Pipes and Filters concurrent Design Pattern using the Concurrency and Coordination Runtime

using Microsoft.Ccr.Core;
using System.Collections.Generic;

namespace WorkCardDemo
{
    class DemuxStateMachine
               : DemuxMessageFilter<StateMachineFilter,
                                    PortSet<CorrectPinMessage, WrongPinMessage>,
                                    PortSet<OpenAlarm, CloseAlarm>,
                                    string>
    {
        public DemuxStateMachine()
            : base(new WorkerKeyValueFilter())
        {

        }
    }

    class WorkerKeyValueFilter : IMessageFilter<PortSet<CorrectPinMessage, WrongPinMessage>, Port<KeyValuePair<string, object>>>
    {

        #region IMessageFilter<PortSet<SuccessMessage,FailureMessage>,Port<KeyValuePair<string,object>> Members

        public PortSet<CorrectPinMessage, WrongPinMessage> InputPort { get; private set; }

        public Port<KeyValuePair<string, object>> OutputPort { get; private set; }

        public void Initialize(DispatcherQueue dispatcherQueue, PortSet<CorrectPinMessage, WrongPinMessage> inputPort, Port<KeyValuePair<string, object>> outputPort)
        {
            InputPort = inputPort;
            OutputPort = outputPort;

            Arbiter.Activate(
                dispatcherQueue,
                Arbiter.Interleave(
                        new TeardownReceiverGroup(),
                        new ExclusiveReceiverGroup(),
                        new ConcurrentReceiverGroup(
                            Arbiter.Receive(true, InputPort.P0, HandleMessage),
                            Arbiter.Receive(true, InputPort.P1, HandleMessage))));
        }

        #endregion

        void HandleMessage<T>(T message) where T : WorkCardSwipeMessageBase
        {
            OutputPort.Post(new KeyValuePair<string, object>(message.Name, message));
        }
    }

}