Scaling-out Your Services by Message Bus based WCF Transport Extension – Part 4

News

Shaun, the author of this blog is a semi-geek, clumsy developer, passionate speaker and incapable architect with about 10 years’ experience in .NET and JavaScript. He hopes to prove that software development is art rather than manufacturing. He's into cloud computing platform and technologies (Windows Azure, Amazon and Aliyun) and right now, Shaun is being attracted by JavaScript (Angular.js and Node.js) and he likes it.

Shaun is working at Worktile Inc. as the chief architect for overall design and develop worktile, a web-based collaboration and task management tool, and lesschat, a real-time communication aggregation tool.

My Stats

Posts - 122

Comments - 622

Trackbacks - 0

Post Categories

.NET

<< Scaling-out Your Services by Message Bus based WCF Transport Extension – Part 3 – Service Host and Client Proxy Support | Home | Scaling-out Your Services by Message Bus based WCF Transport Extension – Part 5 – Duplex, Peak Challenge Comes >>

Scaling-out Your Services by Message Bus based WCF Transport Extension – Part 4 – Datagram MEP

In the previous posts we talked about the transport extension that scaling-out our WCF services over a message bus transportation. What we have done is to use request reply MEP as a example, and implement the asynchronous methods that makes our transport support the normal WCF usage – ServiceHost and ChannelFactory on service hosting and client invoking.

Request reply MEP is the most common mode when we are using WCF, but there are two MEPs available in WCF as well: datagram and duplex. In this post I will demonstrate how to implement the datagram MEP in our solution, with some code refectoring.

Datagram MEP

Datagram MEP, also known as the one-way, request-forget or fire-forget MEP. Different from the request reply, in this mode the client will send the request to the server and no need to wait for the reply. This means, on one hand, the client will not be blocked by the server reply. On the other hand, the client will neither know if the server received the message, nor the service works properly. This MEP is widely used in scenario like logging, auditing, etc..

On the client side, the OutputChannel will send the request message to the underlying transportation, while on the server side, the InputChannel will receive them and dispatched by WCF to the service class and related method.

The OutputChannel just sends the message to the message bus, in our case, and will not wait for any response. It will return back immediately. And the InputChannel on the server will receive the message and process the business logic without sending back any reply.

Hence in order to implement the datagram MEP what we need to do is,

Client side, implement a ChannelFactory that returns an OutputChannel, and implement the OutputChannel which can send message to the bus.
Server side, implement a ChannelListener that returns an InputChannel, and implement the InputChannel which can receive message from the bus.

Since we had created a ChannelFactory and ChannelListener for the request-reply mode, and the channels works well with our message bus before, then we will do some code refactoring first.

ChannelBase, ChannelFactoryBase and ChannelListenerBase

We will have three base classes that covers some common procedures for channels, channel factories and channel listeners. The channel base class will just receive the buffer manager, encoder factory and provide them as a protected readonly properties. We will also have a property named ChannelID, which will be used in the duplex MEP in the next post.

In the request channel and reply channel we have a lot of code that convert the WCF message into string and vice versa, which using the buffer manager and encoder. Here we will create two methods for them so that in the future the channels can convert between string and message by just invoking them.

        1: public abstract class MessageBusChannelBase : ChannelBase

       2: {

       3:     private const int CST_MAXBUFFERSIZE = 64 * 1024;

       4:  

       5:     private readonly Guid _id;

       6:     private readonly BufferManager _bufferManager;

       7:     private readonly MessageEncoder _encoder;

       8:  

       9:     protected BufferManager BufferManager

    {

        get

        {

            return _bufferManager;

        }

    }

  

    protected MessageEncoder Encoder

    {

        get

        {

            return _encoder;

        }

    }

  

    public string ChannelID

    {

        get

        {

            return _id.ToString();

        }

    }

  

    protected MessageBusChannelBase(BufferManager bufferManager, MessageEncoderFactory encoder, ChannelManagerBase parent)

        : base(parent)

    {

        _id = Guid.NewGuid();

  

        _bufferManager = bufferManager;

        _encoder = encoder.CreateSessionEncoder();

    }

  

    internal Message GetWcfMessageFromString(string content)

    {

        var raw = Encoding.UTF8.GetBytes(content);

        var data = _bufferManager.TakeBuffer(raw.Length);

        Buffer.BlockCopy(raw, 0, data, 0, raw.Length);

        var buffer = new ArraySegment<byte>(data, 0, raw.Length);

        var message = _encoder.ReadMessage(buffer, _bufferManager);

        return message;

    }

  

    internal string GetStringFromWcfMessage(Message message, EndpointAddress to)

    {

        ArraySegment<byte> buffer;

        string content;

        using (message)

        {

            to.ApplyTo(message);

            buffer = _encoder.WriteMessage(message, CST_MAXBUFFERSIZE, _bufferManager);

        }

        content = Encoding.UTF8.GetString(buffer.Array, buffer.Offset, buffer.Count);

        _bufferManager.ReturnBuffer(buffer.Array);

        return content;

    }

  

    protected override void OnAbort()

    {

    }

  

    protected override void OnClose(TimeSpan timeout)

    {

    }

  

    protected override void OnOpen(TimeSpan timeout)

    {

    }

  

    #region Not Implemented

  

    protected override IAsyncResult OnBeginClose(TimeSpan timeout, AsyncCallback callback, object state)

    {

        throw new NotImplementedException();

    }

  

    protected override IAsyncResult OnBeginOpen(TimeSpan timeout, AsyncCallback callback, object state)

    {

        throw new NotImplementedException();

    }

  

    protected override void OnEndClose(IAsyncResult result)

    {

        throw new NotImplementedException();

    }

  

    protected override void OnEndOpen(IAsyncResult result)

    {

        throw new NotImplementedException();

    }

  

    #endregion

}

The main responsibility of the channel factory and channel listener is to create a related channel object. Each factories and listeners will has very similar logic, except which type of the channel it will create. So we can have two base classes to wrap almost all complexity and just let the sub classes to take the responsible to create the actual channel object.

       1: public abstract class MessageBusChannelFactoryBase<TChannel> : ChannelFactoryBase<TChannel> where TChannel : class, IChannel

       2: {

       3:     private readonly MessageBusTransportBindingElement _transportElement;

       4:     private readonly BufferManager _bufferManager;

       5:     private readonly MessageEncoderFactory _encoder;

       6:  

       7:     private readonly IBus _bus;

       8:  

       9:     public IBus Bus

    {

        get

        {

            return _bus;

        }

    }

  

    protected MessageBusTransportBindingElement TransportElement

    {

        get

        {

            return _transportElement;

        }

    }

  

    protected MessageBusChannelFactoryBase(MessageBusTransportBindingElement transportElement, BindingContext context)

        : base(context.Binding)

    {

        _transportElement = transportElement;

        _bufferManager = BufferManager.CreateBufferManager(transportElement.MaxBufferPoolSize, int.MaxValue);

        var encodingElement = context.Binding.Elements.Find<MessageEncodingBindingElement>();

        if (encodingElement == null)

        {

            _encoder = (new TextMessageEncodingBindingElement()).CreateMessageEncoderFactory();

        }

        else

        {

            _encoder = encodingElement.CreateMessageEncoderFactory();

        }

        _bus = transportElement.Bus;

    }

  

    protected abstract TChannel CreateChannel(

        BufferManager bufferManager, MessageEncoderFactory encoder, EndpointAddress remoteAddress,

        MessageBusChannelFactoryBase<TChannel> parent,

        Uri via,

        IBus bus);

  

    protected override void OnOpen(TimeSpan timeout)

    {

    }

  

    protected override void OnClosed()

    {

        base.OnClosed();

  

        _bufferManager.Clear();

        _bus.Dispose();

    }

  

    protected override TChannel OnCreateChannel(System.ServiceModel.EndpointAddress address, Uri via)

    {

        return CreateChannel(_bufferManager, _encoder, address, this, via, _bus);

    }

  

    protected override IAsyncResult OnBeginOpen(TimeSpan timeout, AsyncCallback callback, object state)

    {

        throw new NotImplementedException();

    }

  

    protected override void OnEndOpen(IAsyncResult result)

    {

        throw new NotImplementedException();

    }

}

       1: public abstract class MessageBusChannelListenerBase<TChannel> : ChannelListenerBase<TChannel> where TChannel : class, IChannel

       2: {

       3:     private readonly MessageBusTransportBindingElement _transportElement;

       4:  

       5:     private readonly BufferManager _bufferManager;

       6:     private readonly MessageEncoderFactory _encoder;

       7:     private readonly string _scheme;

       8:     private readonly Uri _uri;

       9:  

    private readonly IBus _bus;

  

    private readonly InputQueue<TChannel> _channelQueue;

    private readonly object _currentChannelLock;

  

    private TChannel _currentChannel;

  

    public IBus Bus

    {

        get

        {

            return _bus;

        }

    }

  

    public override Uri Uri

    {

        get

        {

            return _uri;

        }

    }

  

    public string Scheme

    {

        get

        {

            return _scheme;

        }

    }

  

    protected MessageBusTransportBindingElement TransportElement

    {

        get

        {

            return _transportElement;

        }

    }

  

    protected MessageBusChannelListenerBase(MessageBusTransportBindingElement transportElement, BindingContext context)

        : base(context.Binding)

    {

        _transportElement = transportElement;

        _bufferManager = BufferManager.CreateBufferManager(transportElement.MaxBufferPoolSize, int.MaxValue);

        var encodingElement = context.Binding.Elements.Find<MessageEncodingBindingElement>();

        if (encodingElement == null)

        {

            _encoder = (new TextMessageEncodingBindingElement()).CreateMessageEncoderFactory();

        }

        else

        {

            _encoder = encodingElement.CreateMessageEncoderFactory();

        }

        _scheme = transportElement.Scheme;

        _uri = new Uri(context.ListenUriBaseAddress, context.ListenUriRelativeAddress);

  

        _bus = transportElement.Bus;

  

        _channelQueue = new InputQueue<TChannel>();

        _currentChannelLock = new object();

        _currentChannel = null;

    }

  

    protected override void OnAbort()

    {

        try

        {

            lock (ThisLock)

            {

                _channelQueue.Close();

            }

        }

        catch { }

    }

  

    protected override void OnClose(TimeSpan timeout)

    {

        try

        {

            lock (ThisLock)

            {

                _channelQueue.Close();

            }

        }

        catch { }

    }

  

    protected override void OnClosed()

    {

        base.OnClosed();

  

        try

        {

            _bufferManager.Clear();

            _bus.Dispose();

        }

        catch { }

    }

  

    private void EnsureChannelAvailable()

    {

        TChannel newChannel = null;

        bool channelCreated = false;

  

        if ((newChannel = _currentChannel) == null)

        {

            lock (_currentChannelLock)

            {

                if ((newChannel = _currentChannel) == null)

                {

                    newChannel = CreateChannel(_bufferManager, _encoder, new EndpointAddress(_uri), this, _bus);

                    newChannel.Closed += new EventHandler(OnChannelClosed);

                    _currentChannel = newChannel;

                    channelCreated = true;

                }

            }

        }

  

        if (channelCreated)

        {

            _channelQueue.EnqueueAndDispatch(newChannel);

        }

    }

  

    protected abstract TChannel CreateChannel(

        BufferManager bufferManager, MessageEncoderFactory encoder, EndpointAddress localAddress,

        MessageBusChannelListenerBase<TChannel> parent,

        IBus bus);

  

    private void OnChannelClosed(object sender, EventArgs e)

    {

        var channel = sender as TChannel;

        lock (_currentChannelLock)

        {

            if (channel == _currentChannel)

            {

                _currentChannel = null;

            }

        }

    }

  

    protected override TChannel OnAcceptChannel(TimeSpan timeout)

    {

        if (!IsDisposed)

        {

            EnsureChannelAvailable();

        }

  

        TChannel channel = null;

        if (_channelQueue.Dequeue(timeout, out channel))

        {

            return channel;

        }

        else

        {

            throw new TimeoutException();

        }

    }

  

    protected override IAsyncResult OnBeginAcceptChannel(TimeSpan timeout, AsyncCallback callback, object state)

    {

        if (!IsDisposed)

        {

            EnsureChannelAvailable();

        }

  

        return _channelQueue.BeginDequeue(timeout, callback, state);

    }

  

    protected override TChannel OnEndAcceptChannel(IAsyncResult result)

    {

        TChannel channel;

        if (_channelQueue.EndDequeue(result, out channel))

        {

            return channel;

        }

        else

        {

            throw new TimeoutException();

        }

    }

  

    protected override void OnOpen(TimeSpan timeout)

    {

    }

  

    protected override IAsyncResult OnBeginWaitForChannel(TimeSpan timeout, AsyncCallback callback, object state)

    {

        throw new NotImplementedException();

    }

  

    protected override bool OnEndWaitForChannel(IAsyncResult result)

    {

        throw new NotImplementedException();

    }

  

    protected override bool OnWaitForChannel(TimeSpan timeout)

    {

        throw new NotImplementedException();

    }

  

    protected override IAsyncResult OnBeginClose(TimeSpan timeout, AsyncCallback callback, object state)

    {

        throw new NotImplementedException();

    }

  

    protected override IAsyncResult OnBeginOpen(TimeSpan timeout, AsyncCallback callback, object state)

    {

        throw new NotImplementedException();

    }

  

    protected override void OnEndClose(IAsyncResult result)

    {

        throw new NotImplementedException();

    }

  

    protected override void OnEndOpen(IAsyncResult result)

    {

        throw new NotImplementedException();

    }

}

As you can see, the ChannelFactoryBase and the ChannelListenerBase I just expose an abstract method which is CreateChannel, that gives a chance to the sub class to instant the actual channel object. After these modification our request and reply part (that had been done in the previous post) could be turned simpler and clearer.

The RequestChannelFactory will be like this.

       1: public class MessageBusRequestChannelFactory : MessageBusChannelFactoryBase<IRequestChannel>

       2: {

       3:     public MessageBusRequestChannelFactory(MessageBusTransportBindingElement transportElement, BindingContext context)

       4:         : base(transportElement, context)

       5:     {

       6:     }

       7:  

       8:     protected override IRequestChannel CreateChannel(

       9:         BufferManager bufferManager, MessageEncoderFactory encoder, EndpointAddress remoteAddress, 

      10:         MessageBusChannelFactoryBase<IRequestChannel> parent, 

      11:         Uri via, 

      12:         IBus bus)

      13:     {

      14:         return new MessageBusRequestChannel(bufferManager, encoder, parent, remoteAddress, via, bus);

      15:     }

      16: }

And this is the ReplyChannelListener.

       1: public class MessageBusReplyChannelListener : MessageBusChannelListenerBase<IReplyChannel>

       2: {

       3:     public MessageBusReplyChannelListener(MessageBusTransportBindingElement transportElement, BindingContext context)

       4:         : base(transportElement, context)

       5:     {

       6:     }

       7:  

       8:     protected override IReplyChannel CreateChannel(

       9:         BufferManager bufferManager, MessageEncoderFactory encoder, EndpointAddress localAddress, 

      10:         MessageBusChannelListenerBase<IReplyChannel> parent, 

      11:         IBus bus)

      12:     {

      13:         return new MessageBusReplyChannel(bufferManager, encoder, localAddress, parent, bus);

      14:     }

      15: }

And below is the RequestChannel and ReplyChannel. I inherited them from the MessageBusChannelBase class and utilized the helper methods to convert between string and message.

       1: public class MessageBusRequestChannel : MessageBusChannelBase, IRequestChannel

       2: {

       3:     private readonly IBus _bus;

       4:     private readonly Uri _via;

       5:     private readonly EndpointAddress _remoteAddress;

       6:     private readonly object _aLock;

       7:  

       8:     public MessageBusRequestChannel(

       9:         BufferManager bufferManager, MessageEncoderFactory encoder, ChannelManagerBase parent,

        EndpointAddress remoteAddress, Uri via, IBus bus)

        : base(bufferManager, encoder, parent)

    {

        _via = via;

        _remoteAddress = remoteAddress;

        _bus = bus;

        _aLock = new object();

    }

  

    public Uri Via

    {

        get

        {

            return _via;

        }

    }

  

    public IAsyncResult BeginRequest(Message message, TimeSpan timeout, AsyncCallback callback, object state)

    {

        throw new NotImplementedException();

    }

  

    public IAsyncResult BeginRequest(Message message, AsyncCallback callback, object state)

    {

        throw new NotImplementedException();

    }

  

    public Message EndRequest(IAsyncResult result)

    {

        throw new NotImplementedException();

    }

  

    public System.ServiceModel.EndpointAddress RemoteAddress

    {

        get 

        {

            return _remoteAddress;

        }

    }

  

    public Message Request(Message message, TimeSpan timeout)

    {

        ThrowIfDisposedOrNotOpen();

        lock (_aLock)

        {

            // unbox the message into string that will be sent into the bus

            var content = GetStringFromWcfMessage(message,_remoteAddress);

            // send the message into bus

            var busMsgId = _bus.SendRequest(content, true, null);

            // waiting for the reply message arrive from the bus

            var replyMsg = _bus.Receive(false, busMsgId);

            // box the message from the bus message content and return back

            var reply = GetWcfMessageFromString(replyMsg.Content);

            return reply;

        }

    }

  

    public Message Request(Message message)

    {

        return Request(message, DefaultSendTimeout);

    }

}

       1: public class MessageBusReplyChannel : MessageBusChannelBase, IReplyChannel

       2: {

       3:     private readonly EndpointAddress _localAddress;

       4:     private readonly object _aLock;

       5:  

       6:     private readonly IBus _bus;

       7:  

       8:     private delegate bool TryReceiveRequestDelegate(TimeSpan timeout, out RequestContext context);

       9:     private TryReceiveRequestDelegate _tryReceiveRequestDelegate;

  

    public MessageBusReplyChannel(

        BufferManager bufferManager, MessageEncoderFactory encoder, EndpointAddress localAddress,

        ChannelManagerBase parent,

        IBus bus)

        : base(bufferManager, encoder, parent)

    {

        _localAddress = localAddress;

        _bus = bus;

        _aLock = new object();

  

        _tryReceiveRequestDelegate = (TimeSpan t, out RequestContext rc) =>

        {

            rc = null;

            // receive the request message from the bus

            var busMsg = _bus.Receive(true, null);

            // box the wcf message

            var message = GetWcfMessageFromString(busMsg.Content);

            // initialize the request context and return

            rc = new MessageBusRequestContext(message, this, _localAddress, _bus, busMsg.MessageID);

            return true;

        };

    }

  

    public System.ServiceModel.EndpointAddress LocalAddress

    {

        get

        {

            return _localAddress;

        }

    }

  

    public bool WaitForRequest(TimeSpan timeout)

    {

        return true;

    }

  

    public RequestContext ReceiveRequest()

    {

        return ReceiveRequest(DefaultReceiveTimeout);

    }

  

    public RequestContext ReceiveRequest(TimeSpan timeout)

    {

        ThrowIfDisposedOrNotOpen();

        lock (_aLock)

        {

            // receive the request message from the bus

            var busMsg = _bus.Receive(true, null);

            // box the wcf message

            var message = GetWcfMessageFromString(busMsg.Content);

            // initialize the request context and return

            return new MessageBusRequestContext(message, this, _localAddress, _bus, busMsg.MessageID);

        }

    }

  

    public IAsyncResult BeginTryReceiveRequest(TimeSpan timeout, AsyncCallback callback, object state)

    {

        RequestContext context;

        return _tryReceiveRequestDelegate.BeginInvoke(timeout, out context, callback, state);

    }

  

    public bool EndTryReceiveRequest(IAsyncResult result, out RequestContext context)

    {

        var ret = _tryReceiveRequestDelegate.EndInvoke(out context, result);

        return ret;

    }

  

    public IAsyncResult BeginReceiveRequest(TimeSpan timeout, AsyncCallback callback, object state)

    {

        throw new NotImplementedException();

    }

  

    public IAsyncResult BeginReceiveRequest(AsyncCallback callback, object state)

    {

        throw new NotImplementedException();

    }

  

    public IAsyncResult BeginWaitForRequest(TimeSpan timeout, AsyncCallback callback, object state)

    {

        throw new NotImplementedException();

    }

  

    public RequestContext EndReceiveRequest(IAsyncResult result)

    {

        throw new NotImplementedException();

    }

  

    public bool EndWaitForRequest(IAsyncResult result)

    {

        throw new NotImplementedException();

    }

  

    public bool TryReceiveRequest(TimeSpan timeout, out RequestContext context)

    {

        throw new NotImplementedException();

    }

}

Also a little bit changes could be made in the RequestContext as well, which utilizes the message convert methods as well.

       1: public class MessageBusRequestContext : RequestContext

       2: {

       3:     private bool _aborted;

       4:     private readonly Message _message;

       5:     private readonly MessageBusReplyChannel _parent;

       6:     private readonly EndpointAddress _address;

       7:     private readonly object _aLock;

       8:     private readonly string _busMessageId;

       9:     private readonly IBus _bus;

  

    private CommunicationState _state;

  

    public MessageBusRequestContext(

        Message message, MessageBusReplyChannel parent,

        EndpointAddress address,

        IBus bus,

        string relatedTo)

    {

        _aborted = false;

        _parent = parent;

        _message = message;

        _address = address;

        _busMessageId = relatedTo;

        _bus = bus;

  

        _aLock = new object();

        _state = CommunicationState.Opened;

    }

  

    public override void Abort()

    {

        lock (_aLock)

        {

            if (_aborted)

            {

                return;

            }

            _aborted = true;

            _state = CommunicationState.Faulted;

        }

    }

  

    public override IAsyncResult BeginReply(Message message, TimeSpan timeout, AsyncCallback callback, object state)

    {

        throw new NotImplementedException();

    }

  

    public override IAsyncResult BeginReply(Message message, AsyncCallback callback, object state)

    {

        throw new NotImplementedException();

    }

  

    public override void Close(TimeSpan timeout)

    {

        lock (_aLock)

        {

            _state = CommunicationState.Closed;

        }

    }

  

    public override void Close()

    {

        Close(TimeSpan.MaxValue);

    }

  

    public override void EndReply(IAsyncResult result)

    {

        throw new NotImplementedException();

    }

  

    public override void Reply(Message message, TimeSpan timeout)

    {

        // unbox the reply message to string

        var content = _parent.GetStringFromWcfMessage(message, _address);

        // send the reply into bus

        _bus.SendReply(content, false, _busMessageId);

    }

  

    public override void Reply(Message message)

    {

        Reply(message, TimeSpan.MaxValue);

    }

  

    public override Message RequestMessage

    {

        get

        {

            return _message;

        }

    }

}

Channel Factory, Listener and Channels for Datagram

After the refactoring it would be easier to create the channel factory, listener and channels for datagram MEP. First we will create the client side OutputChannelFactory, which just need to implement the CreateChannel from its base class (ChannelFactoryBase) and instant an IOutputChannel object.

       1: public class MessageBusOutputChannelFactory : MessageBusChannelFactoryBase<IOutputChannel>

       2: {

       3:     public MessageBusOutputChannelFactory(MessageBusTransportBindingElement transportElement, BindingContext context)

       4:         : base(transportElement, context)

       5:     {

       6:     }

       7:  

       8:     protected override IOutputChannel CreateChannel(

       9:         BufferManager bufferManager, MessageEncoderFactory encoder, EndpointAddress remoteAddress,

      10:         MessageBusChannelFactoryBase<IOutputChannel> parent,

      11:         Uri via,

      12:         IBus bus)

      13:     {

      14:         return new MessageBusOutputChannel(bufferManager, encoder, parent, remoteAddress, via, bus);

      15:     }

      16: }

The MessageBusOutputChannel class we are going to create will be inherited from the MessageBusChannelBase as well, and will implement the IOutputChannel interface. There’s only one important method we should implement which is Send. This method will be invoked when the client send the message to the server to do something. You will notice that this method doesn’t have any return value, which means, as the datagram MEP defined, the client should not care about whether the service will process its message.

Since we have had the convert method in the channel base class, we can simply convert the message into string and pass it to the underlying message bus.

       1: public virtual void Send(Message message, TimeSpan timeout)

       2: {

       3:     var content = GetStringFromWcfMessage(message, RemoteAddress);

       4:     _bus.SendRequest(content, true, ChannelID, null);

       5: }

The full code of the output channel is

       1: public class MessageBusOutputChannel : MessageBusChannelBase, IOutputChannel

       2: {

       3:     private readonly IBus _bus;

       4:     private readonly Uri _via;

       5:     private readonly EndpointAddress _remoteAddress;

       6:  

       7:     public MessageBusOutputChannel(

       8:         BufferManager bufferManager, MessageEncoderFactory encoder, ChannelManagerBase parent,

       9:         EndpointAddress remoteAddress,

        Uri via, IBus bus)

        : base(bufferManager, encoder, parent)

    {

        _bus = bus;

        _via = via;

        _remoteAddress = remoteAddress;

    }

  

    public Uri Via

    {

        get

        {

            return _via;

        }

    }

  

    public EndpointAddress RemoteAddress

    {

        get

        {

            return _remoteAddress;

        }

    }

  

    public IAsyncResult BeginSend(Message message, TimeSpan timeout, AsyncCallback callback, object state)

    {

        throw new NotImplementedException();

    }

  

    public IAsyncResult BeginSend(Message message, AsyncCallback callback, object state)

    {

        throw new NotImplementedException();

    }

  

    public void EndSend(IAsyncResult result)

    {

        throw new NotImplementedException();

    }

  

    public virtual void Send(Message message, TimeSpan timeout)

    {

        var content = GetStringFromWcfMessage(message, RemoteAddress);

        _bus.SendRequest(content, true, ChannelID, null);

    }

  

    public void Send(Message message)

    {

        Send(message, DefaultSendTimeout);

    }

}

On the server side, similarly we need to create the ChannelListener from the base class and just need to implement the code to create the relevant InputChannel instance.

       1: public class MessageBusInputChannelListener : MessageBusChannelListenerBase<IInputChannel>

       2: {

       3:     public MessageBusInputChannelListener(MessageBusTransportBindingElement transportElement, BindingContext context)

       4:         : base(transportElement, context)

       5:     {

       6:     }

       7:  

       8:     protected override IInputChannel CreateChannel(

       9:         BufferManager bufferManager, MessageEncoderFactory encoder, EndpointAddress localAddress,

      10:         MessageBusChannelListenerBase<IInputChannel> parent, 

      11:         IBus bus)

      12:     {

      13:         return new MessageBusInputChannel(bufferManager, encoder, parent, localAddress, bus);

      14:     }

      15: }

The InputChannel need to receive the message sent from the client (InputChannel), by its BeginTryReceive, EndTryReceive, BeginReceive and EndReceive methods. We will also create two delegates for them and use the asynchronous invoke methods.

       1: public class MessageBusInputChannel : MessageBusChannelBase, IInputChannel

       2: {

       3:     private readonly IBus _bus;

       4:     private readonly EndpointAddress _localAddress;

       5:     private readonly object _aLock;

       6:  

       7:     private delegate bool TryReceiveDelegate(TimeSpan timeout, out Message message);

       8:     private TryReceiveDelegate _tryReceiveDelegate;

       9:  

    private delegate Message ReceiveDelegate(TimeSpan timeout);

    private ReceiveDelegate _receiveDelegate;

  

    public MessageBusInputChannel(

        BufferManager bufferManager, MessageEncoderFactory encoder, ChannelManagerBase parent,

        EndpointAddress localAddress,

        IBus bus)

        : base(bufferManager, encoder, parent)

    {

        _localAddress = localAddress;

        _bus = bus;

        _aLock = new object();

  

        _tryReceiveDelegate = (TimeSpan timeout, out Message message) =>

        {

            message = null;

            try

            {

                var requestMessage = _bus.Receive(true, null);

                if (requestMessage != null)

                {

                    message = GetWcfMessageFromString(requestMessage.Content);

                    OnTryReceive(requestMessage);

                }

            }

            catch (Exception ex)

            {

                throw new CommunicationException(ex.Message, ex);

            }

            return true;

        };

  

        _receiveDelegate = (TimeSpan timeout) =>

        {

            var requestMessage = _bus.Receive(false, ChannelID);

            return GetWcfMessageFromString(requestMessage.Content);

        };

    }

  

    public System.ServiceModel.EndpointAddress LocalAddress

    {

        get

        {

            return _localAddress;

        }

    }

  

    public IAsyncResult BeginReceive(TimeSpan timeout, AsyncCallback callback, object state)

    {

        return _receiveDelegate.BeginInvoke(timeout, callback, state);

    }

  

    public IAsyncResult BeginReceive(AsyncCallback callback, object state)

    {

        return BeginReceive(DefaultReceiveTimeout, callback, state);

    }

  

    public IAsyncResult BeginTryReceive(TimeSpan timeout, AsyncCallback callback, object state)

    {

        Message message;

        return _tryReceiveDelegate.BeginInvoke(timeout, out message, callback, state);

    }

  

    public IAsyncResult BeginWaitForMessage(TimeSpan timeout, AsyncCallback callback, object state)

    {

        throw new NotImplementedException();

    }

  

    public Message EndReceive(IAsyncResult result)

    {

        return _receiveDelegate.EndInvoke(result);

    }

  

    public virtual bool EndTryReceive(IAsyncResult result, out Message message)

    {

        var ret = _tryReceiveDelegate.EndInvoke(out message, result);

        return ret;

    }

  

    protected virtual void OnTryReceive(BusMessage message)

    {

    }

  

    public bool EndWaitForMessage(IAsyncResult result)

    {

        throw new NotImplementedException();

    }

  

    public Message Receive(TimeSpan timeout)

    {

        throw new NotImplementedException();

    }

  

    public Message Receive()

    {

        throw new NotImplementedException();

    }

  

    public bool TryReceive(TimeSpan timeout, out Message message)

    {

        throw new NotImplementedException();

    }

  

    public bool WaitForMessage(TimeSpan timeout)

    {

        throw new NotImplementedException();

    }

}

You might notice that we defined a virtual method named OnTryRecevie and invoked it after the delegate got a valid incoming message, but without doing anything. This method will be used in the future in session implement. You can ignore it for now.

Last thing we need to do is to update the transport binding element, which will make our transport support the InputChannel and OutputChannel, and return the related ChannelFactory and ChannelListener.

       1: public override bool CanBuildChannelFactory<TChannel>(BindingContext context)

       2: {

       3:     return typeof(TChannel) == typeof(IRequestChannel) ||

       4:            typeof(TChannel) == typeof(IOutputChannel);

       5: }

       6:  

       7: public override bool CanBuildChannelListener<TChannel>(BindingContext context)

       8: {

       9:     return typeof(TChannel) == typeof(IReplyChannel) ||

           typeof(TChannel) == typeof(IInputChannel);

}

  

public override IChannelFactory<TChannel> BuildChannelFactory<TChannel>(BindingContext context)

{

    if (context == null)

    {

        throw new ArgumentNullException("context");

    }

    if (!CanBuildChannelFactory<TChannel>(context))

    {

        throw new ArgumentException(string.Format("Unsupported channel type: {0} for the channel factory.", typeof(TChannel).Name));

    }

  

    if (typeof(TChannel) == typeof(IRequestChannel))

    {

        return (IChannelFactory<TChannel>)(object)new MessageBusRequestChannelFactory(this, context);

    }

    else if (typeof(TChannel) == typeof(IOutputChannel))

    {

        return (IChannelFactory<TChannel>)(object)new MessageBusOutputChannelFactory(this, context);

    }

    else

    {

        throw new ArgumentException(string.Format("Unsupported channel type: {0} for the channel listener.", typeof(TChannel).Name));

    }

  

}

  

public override IChannelListener<TChannel> BuildChannelListener<TChannel>(BindingContext context)

{

    if (context == null)

    {

        throw new ArgumentNullException("context");

    }

    if (!CanBuildChannelListener<TChannel>(context))

    {

        throw new ArgumentException(string.Format("Unsupported channel type: {0} for the channel listener.", typeof(TChannel).Name));

    }

  

    if (typeof(TChannel) == typeof(IReplyChannel))

    {

        return (IChannelListener<TChannel>)(object)new MessageBusReplyChannelListener(this, context);

    }

    else if (typeof(TChannel) == typeof(IInputChannel))

    {

        return (IChannelListener<TChannel>)(object)new MessageBusInputChannelListener(this, context);

    }

    else

    {

        throw new ArgumentException(string.Format("Unsupported channel type: {0} for the channel listener.", typeof(TChannel).Name));

    }

}

Now let’s test our datagram operation. Open the test console project and create a new service contract with one method which has the IsOneWay = true defined.

       1: [ServiceContract(Namespace = "http://wcf.shaunxu.me/")]

       2: public interface ISampleService

       3: {

       4:     [OperationContract(IsOneWay = true)]

       5:     void Ping();

       6: }

       7:  

       8: public class SampleService : ISampleService

       9: {

      10:     public void Ping()

      11:     {

      12:         Console.WriteLine("Service {0}: PONG!", OperationContext.Current.Host.GetHashCode());

      13:     }

      14: }

And update the console main function as well.

       1: static void Main(string[] args)

       2: {

       3:     var bus = new InProcMessageBus();

       4:     var address = "net.bus://localhost/sample";

       5:  

       6:     // establish the services

       7:     var host1 = EstablishServiceHost<ISampleService, SampleService>(bus, address);

       8:     var host2 = EstablishServiceHost<ISampleService, SampleService>(bus, address);

       9:     var host3 = EstablishServiceHost<ISampleService, SampleService>(bus, address);

  

    // establish the client

    var cliBinding = new MessageBusTransportBinding(bus);

    var factory = new ChannelFactory<ISampleService>(cliBinding, address);

    factory.Opened += (sender, e) =>

        {

            Console.WriteLine("Client connected to {0}", factory.Endpoint.ListenUri);

        };

    var proxy = factory.CreateChannel();

  

    // invoke the service

    using (proxy as IDisposable)

    {

        Console.WriteLine("Press 'p' to ping the service, otherwise to exit.");

        var content = Console.ReadLine();

        while (string.Compare(content, "p", true) == 0)

        {

            proxy.Ping();

            content = Console.ReadLine();

        }

    }

}

Again, I’m using the in process message bus so the multi-instances service will be hosted on threads instead of processes and machines. When we invoked the service many times we can see the datagram message send across to the three services.

IsOneWay = ture Always Means Input-Output Channel?

Let’s modified our test service contract and service implementation a little bit, and to see what happen. Let’s add another method in the contract with the IsOneWay = false, as well as the implementation class.

       1: [ServiceContract(Namespace = "http://wcf.shaunxu.me/")]

       2: public interface ISampleService

       3: {

       4:     [OperationContract(IsOneWay = true)]

       5:     void Ping();

       6:  

       7:     [OperationContract(IsOneWay = false)]

       8:     string Reverse(string content);

       9: }

      10:  

      11: public class SampleService : ISampleService

      12: {

      13:     public void Ping()

      14:     {

      15:         Console.WriteLine("Service {0}: PONG!", OperationContext.Current.Host.GetHashCode());

      16:     }

      17:  

      18:     public string Reverse(string content)

      19:     {

      20:         var result = new string(content.Reverse().ToArray());

      21:         Console.WriteLine("Service {0}: Reverse {1} => {2}", OperationContext.Current.Host.GetHashCode(), content, result);

      22:         return result;

      23:     }

      24: }

And then F5 our console application and try to invoke the Ping method. Now we found that the client sent the request, but the service method was not invoked and our client was dead.

More interesting, if we debugged the code we will find that when we invoked the Ping service method which is IsOneWay = true, the WCF tried to create a RequestChannel and ReplyChannel to us.

Why I invoked a service method with IsOneWay = true, but the WCF gave me a request and reply Channel? Well the channel the WCF chose is NOT based on the method you are invoking, it is based on the contract definition. The channel selection logic is complex but to be simple, WCF will select the channels which just satisfied with the requirement of your service contract. Kenny Wolf have a nice post that roughly summarized this logic.

In our case, we have a method marked IsOneWay = true and another marked IsOneWay = false. So the WCF will chose the request reply channels even though we are using the method that IsOneWay = true. The problem is that, when using request reply channel handles the datagram message, the reply message from WCF will always be null. This will cause the exception in our RequestContext.Reply method, since we need to send the reply message back to the client but now it’s null.

Thanks Jiang Jinnan, who helped me a lot to figure this problem out and the solution suggestion. For more information about this WCF MVP you can visit his blog (in Chinese).

When we knew the root caution, the solution would be very simple. In the RequestContext.Reply method we will check if the message is null. If yes, that means this is a datagram channel shape and we need to send a blank message back to the client to acknowledge and make sure the process is running.

       1: public override void Reply(Message message, TimeSpan timeout)

       2: {

       3:     if (message == null)

       4:     {

       5:         // this means this is a one way message

       6:         // we just need to send a blank message back to the client to acknowledge it

       7:         _bus.SendReply(string.Empty, false, _busMessageId);

       8:     }

       9:     else

      10:     {

      11:         // unbox the reply message to string

      12:         var content = _parent.GetStringFromWcfMessage(message, _address);

      13:         // send the reply into bus

      14:         _bus.SendReply(content, false, _busMessageId);

      15:     }

      16: }

And in the RequestChannel.Request we also need to check whether the reply message content is blank. If yes we will do nothing since we know that this is a datagram channel acknowledge message from the server side.

       1: public Message Request(Message message, TimeSpan timeout)

       2: {

       3:     ThrowIfDisposedOrNotOpen();

       4:     lock (_aLock)

       5:     {

       6:         // unbox the message into string that will be sent into the bus

       7:         var content = GetStringFromWcfMessage(message,_remoteAddress);

       8:         // send the message into bus

       9:         var busMsgId = _bus.SendRequest(content, true, null);

      10:         // waiting for the reply message arrive from the bus

      11:         var replyMsg = _bus.Receive(false, busMsgId);

      12:         if (string.IsNullOrWhiteSpace(replyMsg.Content))

      13:         {

      14:             // this means this is a one way channel acknowledge from server

      15:             // we just return null and do nothing

      16:             return null;

      17:         }

      18:         else

      19:         {

      20:             // box the message from the bus message content and return back

      21:             var reply = GetWcfMessageFromString(replyMsg.Content);

      22:             return reply;

      23:         }

      24:     }

      25: }

Now let’s execute our test console you can see the IsOneWay = ture method was invoked correctly.

Summary

In this post I implemented the datagram MEP channel factory, listener and the related input output channels. I also explained a potential problem if we have IsOneWay = true methods and IsOneWay = false methods mixed in one service contract, and how to solve it.

We have finished two MEPs in our transport extension, request reply and datagram. And both of them supports the service side scaling-out, as you can see our request can be handled by multiple services, based on the pulling approach.

In the next post I will demonstrate the most complexity and difficulty part, the duplex mode. I will implement the duplex channels over our message bus infrastructure with the server side scalability and sticky to the same client instance as well.

You can download the source code here.

Hope this helps,

Shaun

All documents and related graphics, codes are provided "AS IS" without warranty of any kind.
Copyright © Shaun Ziyan Xu. This work is licensed under the Creative Commons License.

Share This Post:
Short Url: http://wblo.gs/cmf

Posted On Friday, March 23, 2012 3:07 PM
Filed Under [ Architecture Cloud Computing ]

Comments

No comments posted yet.

Title:
Name:
Email:
Comment:
Verification:
	Remember Me?

Shaun Xu

The Sheep-Pen of the Shaun

News

My Stats

Tag Cloud

Recent Comments

Recent Posts

Archives

Post Categories

.NET

Scaling-out Your Services by Message Bus based WCF Transport Extension – Part 4 – Datagram MEP

Datagram MEP

ChannelBase, ChannelFactoryBase and ChannelListenerBase

Channel Factory, Listener and Channels for Datagram

IsOneWay = ture Always Means Input-Output Channel?

Summary

Posted On Friday, March 23, 2012 3:07 PM
Filed Under [ Architecture Cloud Computing ]

Comments

Shaun Xu

The Sheep-Pen of the Shaun

News

My Stats

Tag Cloud

Recent Comments

Recent Posts

Archives

Post Categories

.NET

Scaling-out Your Services by Message Bus based WCF Transport Extension – Part 4 – Datagram MEP

Datagram MEP

ChannelBase, ChannelFactoryBase and ChannelListenerBase

Channel Factory, Listener and Channels for Datagram

IsOneWay = ture Always Means Input-Output Channel?

Summary

Posted On Friday, March 23, 2012 3:07 PM Filed Under [ Architecture Cloud Computing ]

Comments

Posted On Friday, March 23, 2012 3:07 PM
Filed Under [ Architecture Cloud Computing ]