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

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 2 – Frist Trying Request-Reply Mode | Home | Scaling-out Your Services by Message Bus based WCF Transport Extension – Part 4 – Datagram MEP >>

Scaling-out Your Services by Message Bus based WCF Transport Extension – Part 3 – Service Host and Client Proxy Support

In the previous post I demonstrated how to implement a very basic transport extension over an in memory message bus that supports request reply MEP. At the end of that post I created a console application and establish the service and client directly through the channel listener (on the server side) and the channel factory (on the client side). But this is not the WCF usage that we are familiar with. If you have been using the WCF for a while the common pattern is to create a ServiceHost on the server side and a ChannelFactory<T> on the client side. Then we will using the ChannelFactory<T>.CreateChannel to get an instance of the service contract interface so we can invoke the service remotely as if it’s locally.

Using the “Add Service Reference” in Visual Studio or the “svcutil” is the same way as using the ChannelFactory<T> on the client side. In fact the Visual Studio will call the “svcutil” and generate a client proxy to us, which is similar as the service contract one but will some connection functions.

Based on the transport extension we had finished so far, we can using the ServiceHost and ChannelFactory at once. But since we just implemented the synchronized methods, for example the ReceiveRequest. We must indicate to WCF infrastructure to use the synchronized methods.

        1: <endpointBehaviors>

       2:   <behavior name="">

       3:     <synchronousReceive />

       4:   </behavior>

       5: </endpointBehaviors>

Now I will describe how to amend our transport extension to support the asynchronous mode channel creation and message operation so that we can use the ServiceHost and ClientFactory<T>, which is the way that we are familiar with. Again, this is not mandatory for transport extension since it works well even though only support the synchronous mode. But in order to maximize the performance and usability, this is very important.

Client: ChannelFactory and RequestChannel

On the client side the transport binding element will create the channel factory and it will initialize a request channel object in its CreateChannel method. All of them are invoked synchronously and there’s no related asynchronous methods available. This means in WCF, the client operation will always be invoked synchronously.

Although we can generate a client proxy class from the “svcutil” with the asynchronously methods, this will only affect on the calling thread. The channel factory and channel will still be created and invoked synchronously in its thread.

It’s the same on the client side channel, the RequestChannel in our case. It only support the synchronously method to send the request message. So on the client side we don’t need to modify anything.

Server: ChannelListener

On the server side it would be a little bit complex. In WCF one service could be able to handle multiple requests at the same time by using the different threads. This means on the server side besides the synchronous methods we must implement the asynchronous ones to support the concurrency mode.

As we know the ChannelListener will create a server side channel when any request came to. In our current code we just initialized a new ReplyChannel instance and returned in the OnAcceptChannel method. But if we are going to use the ServiceHost class to establish the service we must implement its asynchronous method pair, which are OnBeginAcceptChannel and OnEndAcceptChannel. One of the implementation is very simple, we can create a new delegate that point to the OnAcceptChannel method we had done before, and use the asynchronous invoke methods to implement the OnBeginAcceptChannel and the OnEndAcceptChannel. We will define a delegation and assign the OnAcceptChannel to it in the class constructor, and use the BeginInvoke method in the OnBeginAcceptChannel while EndInvoke in the OnEndAcceptChannel method.

       1: public class MessageBusReplyChannelListener : ChannelListenerBase<IReplyChannel>

       2: {

       3:     ... ...

       4:  

       5:     private delegate IReplyChannel AcceptChannelDelegate(TimeSpan timeout);

       6:     private AcceptChannelDelegate _acceptChannelDelegate;

       7:  

       8:     ... ... 

       9:  

    public MessageBusReplyChannelListener(MessageBusTransportBindingElement transportElement, BindingContext context)

        : base(context.Binding)

    {

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

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

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

        _bus = transportElement.Bus;

  

        _acceptChannelDelegate = OnAcceptChannel;

    }

  

    ... ...

  

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

    {

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

    }

  

    protected override IReplyChannel OnEndAcceptChannel(IAsyncResult result)

    {

        return _acceptChannelDelegate.EndInvoke(result);

    }

  

    ... ...

}

This is OK and works when using the ServiceHost, and we will see that we are going to use the same way when amend the ReplyChannel part. But this is not the best way of doing the ChannelListener. It’s because if we are doing like this all incoming request will cause the service to try to create a new ReplyChannel, regardless if the server have the enough computing and memory resource available. In WCF we can define the throughput on a server so all client request will be queued on the server, and the server will create the channels based on its availability. Microsoft gives us a very good class to implement this queue with the throughput controlling, as well as the asynchronous enqueue, dequeue and dispatch methods, which the name is InputQueue<T>.

InputQueue<T> was not provided within the BCL of the .NET Framework, although it’s very useful. But there are many ways to get it. One is to use some decompile tool, such as the ILSpy, to get its source code from the System.ServiceModel.dll. It’s an internal class inside the System.ServiceModel.Chanels namespace.

Alternatively you can download the WCF samples and after installed, you can find this class in the folder \WCF\Extensibility\Transport\Udp\CS\UdpTransport\InputQueue.cs. Below this is source code of this class I’m going to use in our example.

       1: // ItemDequeuedCallback is called as an item is dequeued from the InputQueue.  The 

       2: // InputQueue lock is not held during the callback.  However, the user code is

       3: // not notified of the item being available until the callback returns.  If you

       4: // are not sure if the callback blocks for a long time, then first call 

       5: // IOThreadScheduler.ScheduleCallback to get to a "safe" thread.

       6: delegate void ItemDequeuedCallback();

       7:  

       8: /// <summary>

       9: /// Handles asynchronous interactions between producers and consumers. 

/// Producers can dispatch available data to the input queue, 

/// where it is dispatched to a waiting consumer or stored until a

/// consumer becomes available. Consumers can synchronously or asynchronously

/// request data from the queue, which is returned when data becomes

/// available.

/// </summary>

/// <typeparam name="T">The concrete type of the consumer objects that are waiting for data.</typeparam>

internal class InputQueue<T> : IDisposable where T : class

{

    //Stores items that are waiting to be accessed.

    private ItemQueue itemQueue;

  

    //Each IQueueReader represents some consumer that is waiting for

    //items to appear in the queue. The readerQueue stores them

    //in an ordered list so consumers get serviced in a FIFO manner.

    private Queue<IQueueReader> readerQueue;

  

    //Each IQueueWaiter represents some waiter that is waiting for

    //items to appear in the queue.  When any item appears, all

    //waiters are signaled.

    private List<IQueueWaiter> waiterList;

  

    private static WaitCallback onInvokeDequeuedCallback;

    private static WaitCallback onDispatchCallback;

    private static WaitCallback completeOutstandingReadersCallback;

    private static WaitCallback completeWaitersFalseCallback;

    private static WaitCallback completeWaitersTrueCallback;

  

    //Represents the current state of the InputQueue.

    //as it transitions through its lifecycle.

    QueueState queueState;

    enum QueueState

    {

        Open,

        Shutdown,

        Closed

    }

  

    public InputQueue()

    {

        this.itemQueue = new ItemQueue();

        this.readerQueue = new Queue<IQueueReader>();

        this.waiterList = new List<IQueueWaiter>();

        this.queueState = QueueState.Open;

    }

  

    public int PendingCount

    {

        get

        {

            lock (ThisLock)

            {

                return itemQueue.ItemCount;

            }

        }

    }

  

    object ThisLock

    {

        get { return itemQueue; }

    }

  

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

    {

        Item item = default(Item);

  

        lock (ThisLock)

        {

            if (queueState == QueueState.Open)

            {

                if (itemQueue.HasAvailableItem)

                {

                    item = itemQueue.DequeueAvailableItem();

                }

                else

                {

                    AsyncQueueReader reader = new AsyncQueueReader(this, timeout, callback, state);

                    readerQueue.Enqueue(reader);

                    return reader;

                }

            }

            else if (queueState == QueueState.Shutdown)

            {

                if (itemQueue.HasAvailableItem)

                {

                    item = itemQueue.DequeueAvailableItem();

                }

                else if (itemQueue.HasAnyItem)

                {

                    AsyncQueueReader reader = new AsyncQueueReader(this, timeout, callback, state);

                    readerQueue.Enqueue(reader);

                    return reader;

                }

            }

        }

  

        InvokeDequeuedCallback(item.DequeuedCallback);

        return new TypedCompletedAsyncResult<T>(item.GetValue(), callback, state);

    }

  

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

    {

        lock (ThisLock)

        {

            if (queueState == QueueState.Open)

            {

                if (!itemQueue.HasAvailableItem)

                {

                    AsyncQueueWaiter waiter = new AsyncQueueWaiter(timeout, callback, state);

                    waiterList.Add(waiter);

                    return waiter;

                }

            }

            else if (queueState == QueueState.Shutdown)

            {

                if (!itemQueue.HasAvailableItem && itemQueue.HasAnyItem)

                {

                    AsyncQueueWaiter waiter = new AsyncQueueWaiter(timeout, callback, state);

                    waiterList.Add(waiter);

                    return waiter;

                }

            }

        }

  

        return new TypedCompletedAsyncResult<bool>(true, callback, state);

    }

  

    static void CompleteOutstandingReadersCallback(object state)

    {

        IQueueReader[] outstandingReaders = (IQueueReader[])state;

  

        for (int i = 0; i < outstandingReaders.Length; i++)

        {

            outstandingReaders[i].Set(default(Item));

        }

    }

  

    static void CompleteWaitersFalseCallback(object state)

    {

        CompleteWaiters(false, (IQueueWaiter[])state);

    }

  

    static void CompleteWaitersTrueCallback(object state)

    {

        CompleteWaiters(true, (IQueueWaiter[])state);

    }

  

    static void CompleteWaiters(bool itemAvailable, IQueueWaiter[] waiters)

    {

        for (int i = 0; i < waiters.Length; i++)

        {

            waiters[i].Set(itemAvailable);

        }

    }

  

    static void CompleteWaitersLater(bool itemAvailable, IQueueWaiter[] waiters)

    {

        if (itemAvailable)

        {

            if (completeWaitersTrueCallback == null)

                completeWaitersTrueCallback = new WaitCallback(CompleteWaitersTrueCallback);

  

            ThreadPool.QueueUserWorkItem(completeWaitersTrueCallback, waiters);

        }

        else

        {

            if (completeWaitersFalseCallback == null)

                completeWaitersFalseCallback = new WaitCallback(CompleteWaitersFalseCallback);

  

            ThreadPool.QueueUserWorkItem(completeWaitersFalseCallback, waiters);

        }

    }

  

    void GetWaiters(out IQueueWaiter[] waiters)

    {

        if (waiterList.Count > 0)

        {

            waiters = waiterList.ToArray();

            waiterList.Clear();

        }

        else

        {

            waiters = null;

        }

    }

  

    public void Close()

    {

        ((IDisposable)this).Dispose();

    }

  

    public void Shutdown()

    {

        IQueueReader[] outstandingReaders = null;

        lock (ThisLock)

        {

            if (queueState == QueueState.Shutdown)

                return;

  

            if (queueState == QueueState.Closed)

                return;

  

            this.queueState = QueueState.Shutdown;

  

            if (readerQueue.Count > 0 && this.itemQueue.ItemCount == 0)

            {

                outstandingReaders = new IQueueReader[readerQueue.Count];

                readerQueue.CopyTo(outstandingReaders, 0);

                readerQueue.Clear();

            }

        }

  

        if (outstandingReaders != null)

        {

            for (int i = 0; i < outstandingReaders.Length; i++)

            {

                outstandingReaders[i].Set(new Item((Exception)null, null));

            }

        }

    }

  

    public T Dequeue(TimeSpan timeout)

    {

        T value;

  

        if (!this.Dequeue(timeout, out value))

        {

            throw new TimeoutException(string.Format("Dequeue timed out in {0}.", timeout));

        }

  

        return value;

    }

  

    public bool Dequeue(TimeSpan timeout, out T value)

    {

        WaitQueueReader reader = null;

        Item item = new Item();

  

        lock (ThisLock)

        {

            if (queueState == QueueState.Open)

            {

                if (itemQueue.HasAvailableItem)

                {

                    item = itemQueue.DequeueAvailableItem();

                }

                else

                {

                    reader = new WaitQueueReader(this);

                    readerQueue.Enqueue(reader);

                }

            }

            else if (queueState == QueueState.Shutdown)

            {

                if (itemQueue.HasAvailableItem)

                {

                    item = itemQueue.DequeueAvailableItem();

                }

                else if (itemQueue.HasAnyItem)

                {

                    reader = new WaitQueueReader(this);

                    readerQueue.Enqueue(reader);

                }

                else

                {

                    value = default(T);

                    return true;

                }

            }

            else // queueState == QueueState.Closed

            {

                value = default(T);

                return true;

            }

        }

  

        if (reader != null)

        {

            return reader.Wait(timeout, out value);

        }

        else

        {

            InvokeDequeuedCallback(item.DequeuedCallback);

            value = item.GetValue();

            return true;

        }

    }

  

    public void Dispose()

    {

        Dispose(true);

  

        GC.SuppressFinalize(this);

    }

  

    protected void Dispose(bool disposing)

    {

        if (disposing)

        {

            bool dispose = false;

  

            lock (ThisLock)

            {

                if (queueState != QueueState.Closed)

                {

                    queueState = QueueState.Closed;

                    dispose = true;

                }

            }

  

            if (dispose)

            {

                while (readerQueue.Count > 0)

                {

                    IQueueReader reader = readerQueue.Dequeue();

                    reader.Set(default(Item));

                }

  

                while (itemQueue.HasAnyItem)

                {

                    Item item = itemQueue.DequeueAnyItem();

                    item.Dispose();

                    InvokeDequeuedCallback(item.DequeuedCallback);

                }

            }

        }

    }

  

    public void Dispatch()

    {

        IQueueReader reader = null;

        Item item = new Item();

        IQueueReader[] outstandingReaders = null;

        IQueueWaiter[] waiters = null;

        bool itemAvailable = true;

  

        lock (ThisLock)

        {

            itemAvailable = !((queueState == QueueState.Closed) || (queueState == QueueState.Shutdown));

            this.GetWaiters(out waiters);

  

            if (queueState != QueueState.Closed)

            {

                itemQueue.MakePendingItemAvailable();

  

                if (readerQueue.Count > 0)

                {

                    item = itemQueue.DequeueAvailableItem();

                    reader = readerQueue.Dequeue();

  

                    if (queueState == QueueState.Shutdown && readerQueue.Count > 0 && itemQueue.ItemCount == 0)

                    {

                        outstandingReaders = new IQueueReader[readerQueue.Count];

                        readerQueue.CopyTo(outstandingReaders, 0);

                        readerQueue.Clear();

  

                        itemAvailable = false;

                    }

                }

            }

        }

  

        if (outstandingReaders != null)

        {

            if (completeOutstandingReadersCallback == null)

                completeOutstandingReadersCallback = new WaitCallback(CompleteOutstandingReadersCallback);

  

            ThreadPool.QueueUserWorkItem(completeOutstandingReadersCallback, outstandingReaders);

        }

  

        if (waiters != null)

        {

            CompleteWaitersLater(itemAvailable, waiters);

        }

  

        if (reader != null)

        {

            InvokeDequeuedCallback(item.DequeuedCallback);

            reader.Set(item);

        }

    }

  

    //Ends an asynchronous Dequeue operation.

    public T EndDequeue(IAsyncResult result)

    {

        T value;

  

        if (!this.EndDequeue(result, out value))

        {

            throw new TimeoutException("Asynchronous Dequeue operation timed out.");

        }

  

        return value;

    }

  

    public bool EndDequeue(IAsyncResult result, out T value)

    {

        TypedCompletedAsyncResult<T> typedResult = result as TypedCompletedAsyncResult<T>;

  

        if (typedResult != null)

        {

            value = TypedCompletedAsyncResult<T>.End(result);

            return true;

        }

  

        return AsyncQueueReader.End(result, out value);

    }

  

    public bool EndWaitForItem(IAsyncResult result)

    {

        TypedCompletedAsyncResult<bool> typedResult = result as TypedCompletedAsyncResult<bool>;

        if (typedResult != null)

        {

            return TypedCompletedAsyncResult<bool>.End(result);

        }

  

        return AsyncQueueWaiter.End(result);

    }

  

    public void EnqueueAndDispatch(T item)

    {

        EnqueueAndDispatch(item, null);

    }

  

    public void EnqueueAndDispatch(T item, ItemDequeuedCallback dequeuedCallback)

    {

        EnqueueAndDispatch(item, dequeuedCallback, true);

    }

  

    public void EnqueueAndDispatch(Exception exception, ItemDequeuedCallback dequeuedCallback, bool canDispatchOnThisThread)

    {

        Debug.Assert(exception != null, "exception parameter should not be null");

        EnqueueAndDispatch(new Item(exception, dequeuedCallback), canDispatchOnThisThread);

    }

  

    public void EnqueueAndDispatch(T item, ItemDequeuedCallback dequeuedCallback, bool canDispatchOnThisThread)

    {

        Debug.Assert(item != null, "item parameter should not be null");

        EnqueueAndDispatch(new Item(item, dequeuedCallback), canDispatchOnThisThread);

    }

  

    void EnqueueAndDispatch(Item item, bool canDispatchOnThisThread)

    {

        bool disposeItem = false;

        IQueueReader reader = null;

        bool dispatchLater = false;

        IQueueWaiter[] waiters = null;

        bool itemAvailable = true;

  

        lock (ThisLock)

        {

            itemAvailable = !((queueState == QueueState.Closed) || (queueState == QueueState.Shutdown));

            this.GetWaiters(out waiters);

  

            if (queueState == QueueState.Open)

            {

                if (canDispatchOnThisThread)

                {

                    if (readerQueue.Count == 0)

                    {

                        itemQueue.EnqueueAvailableItem(item);

                    }

                    else

                    {

                        reader = readerQueue.Dequeue();

                    }

                }

                else

                {

                    if (readerQueue.Count == 0)

                    {

                        itemQueue.EnqueueAvailableItem(item);

                    }

                    else

                    {

                        itemQueue.EnqueuePendingItem(item);

                        dispatchLater = true;

                    }

                }

            }

            else // queueState == QueueState.Closed || queueState == QueueState.Shutdown

            {

                disposeItem = true;

            }

        }

  

        if (waiters != null)

        {

            if (canDispatchOnThisThread)

            {

                CompleteWaiters(itemAvailable, waiters);

            }

            else

            {

                CompleteWaitersLater(itemAvailable, waiters);

            }

        }

  

        if (reader != null)

        {

            InvokeDequeuedCallback(item.DequeuedCallback);

            reader.Set(item);

        }

  

        if (dispatchLater)

        {

            if (onDispatchCallback == null)

            {

                onDispatchCallback = new WaitCallback(OnDispatchCallback);

            }

  

            ThreadPool.QueueUserWorkItem(onDispatchCallback, this);

        }

        else if (disposeItem)

        {

            InvokeDequeuedCallback(item.DequeuedCallback);

            item.Dispose();

        }

    }

  

    public bool EnqueueWithoutDispatch(T item, ItemDequeuedCallback dequeuedCallback)

    {

        Debug.Assert(item != null, "EnqueueWithoutDispatch: item parameter should not be null");

        return EnqueueWithoutDispatch(new Item(item, dequeuedCallback));

    }

  

    public bool EnqueueWithoutDispatch(Exception exception, ItemDequeuedCallback dequeuedCallback)

    {

        Debug.Assert(exception != null, "EnqueueWithoutDispatch: exception parameter should not be null");

        return EnqueueWithoutDispatch(new Item(exception, dequeuedCallback));

    }

  

    // This does not block, however, Dispatch() must be called later if this function

    // returns true.

    bool EnqueueWithoutDispatch(Item item)

    {

        lock (ThisLock)

        {

            // Open

            if (queueState != QueueState.Closed && queueState != QueueState.Shutdown)

            {

                if (readerQueue.Count == 0)

                {

                    itemQueue.EnqueueAvailableItem(item);

                    return false;

                }

                else

                {

                    itemQueue.EnqueuePendingItem(item);

                    return true;

                }

            }

        }

  

        item.Dispose();

        InvokeDequeuedCallbackLater(item.DequeuedCallback);

        return false;

    }

  

    static void OnDispatchCallback(object state)

    {

        ((InputQueue<T>)state).Dispatch();

    }

  

    static void InvokeDequeuedCallbackLater(ItemDequeuedCallback dequeuedCallback)

    {

        if (dequeuedCallback != null)

        {

            if (onInvokeDequeuedCallback == null)

            {

                onInvokeDequeuedCallback = OnInvokeDequeuedCallback;

            }

  

            ThreadPool.QueueUserWorkItem(onInvokeDequeuedCallback, dequeuedCallback);

        }

    }

  

    static void InvokeDequeuedCallback(ItemDequeuedCallback dequeuedCallback)

    {

        if (dequeuedCallback != null)

        {

            dequeuedCallback();

        }

    }

  

    static void OnInvokeDequeuedCallback(object state)

    {

        ItemDequeuedCallback dequeuedCallback = (ItemDequeuedCallback)state;

        dequeuedCallback();

    }

  

    bool RemoveReader(IQueueReader reader)

    {

        lock (ThisLock)

        {

            if (queueState == QueueState.Open || queueState == QueueState.Shutdown)

            {

                bool removed = false;

  

                for (int i = readerQueue.Count; i > 0; i--)

                {

                    IQueueReader temp = readerQueue.Dequeue();

                    if (Object.ReferenceEquals(temp, reader))

                    {

                        removed = true;

                    }

                    else

                    {

                        readerQueue.Enqueue(temp);

                    }

                }

  

                return removed;

            }

        }

  

        return false;

    }

  

    public bool WaitForItem(TimeSpan timeout)

    {

        WaitQueueWaiter waiter = null;

        bool itemAvailable = false;

  

        lock (ThisLock)

        {

            if (queueState == QueueState.Open)

            {

                if (itemQueue.HasAvailableItem)

                {

                    itemAvailable = true;

                }

                else

                {

                    waiter = new WaitQueueWaiter();

                    waiterList.Add(waiter);

                }

            }

            else if (queueState == QueueState.Shutdown)

            {

                if (itemQueue.HasAvailableItem)

                {

                    itemAvailable = true;

                }

                else if (itemQueue.HasAnyItem)

                {

                    waiter = new WaitQueueWaiter();

                    waiterList.Add(waiter);

                }

                else

                {

                    return false;

                }

            }

            else // queueState == QueueState.Closed

            {

                return true;

            }

        }

  

        if (waiter != null)

        {

            return waiter.Wait(timeout);

        }

        else

        {

            return itemAvailable;

        }

    }

  

    interface IQueueReader

    {

        void Set(Item item);

    }

  

    interface IQueueWaiter

    {

        void Set(bool itemAvailable);

    }

  

    class WaitQueueReader : IQueueReader

    {

        Exception exception;

        InputQueue<T> inputQueue;

        T item;

        ManualResetEvent waitEvent;

        object thisLock = new object();

  

        public WaitQueueReader(InputQueue<T> inputQueue)

        {

            this.inputQueue = inputQueue;

            waitEvent = new ManualResetEvent(false);

        }

  

        object ThisLock

        {

            get

            {

                return this.thisLock;

            }

        }

  

        public void Set(Item item)

        {

            lock (ThisLock)

            {

                Debug.Assert(this.item == null, "InputQueue.WaitQueueReader.Set: (this.item == null)");

                Debug.Assert(this.exception == null, "InputQueue.WaitQueueReader.Set: (this.exception == null)");

  

                this.exception = item.Exception;

                this.item = item.Value;

                waitEvent.Set();

            }

        }

  

        public bool Wait(TimeSpan timeout, out T value)

        {

            bool isSafeToClose = false;

            try

            {

                if (timeout == TimeSpan.MaxValue)

                {

                    waitEvent.WaitOne();

                }

                else if (!waitEvent.WaitOne(timeout, false))

                {

                    if (this.inputQueue.RemoveReader(this))

                    {

                        value = default(T);

                        isSafeToClose = true;

                        return false;

                    }

                    else

                    {

                        waitEvent.WaitOne();

                    }

                }

  

                isSafeToClose = true;

            }

            finally

            {

                if (isSafeToClose)

                {

                    waitEvent.Close();

                }

            }

  

            value = item;

            return true;

        }

    }

  

    class AsyncQueueReader : AsyncResult, IQueueReader

    {

        static TimerCallback timerCallback = new TimerCallback(AsyncQueueReader.TimerCallback);

  

        bool expired;

        InputQueue<T> inputQueue;

        T item;

        Timer timer;

  

        public AsyncQueueReader(InputQueue<T> inputQueue, TimeSpan timeout, AsyncCallback callback, object state)

            : base(callback, state)

        {

            this.inputQueue = inputQueue;

            if (timeout != TimeSpan.MaxValue)

            {

                this.timer = new Timer(timerCallback, this, timeout, TimeSpan.FromMilliseconds(-1));

            }

        }

  

        public static bool End(IAsyncResult result, out T value)

        {

            AsyncQueueReader readerResult = AsyncResult.End<AsyncQueueReader>(result);

  

            if (readerResult.expired)

            {

                value = default(T);

                return false;

            }

            else

            {

                value = readerResult.item;

                return true;

            }

        }

  

        static void TimerCallback(object state)

        {

            AsyncQueueReader thisPtr = (AsyncQueueReader)state;

            if (thisPtr.inputQueue.RemoveReader(thisPtr))

            {

                thisPtr.expired = true;

                thisPtr.Complete(false);

            }

        }

  

        public void Set(Item item)

        {

            this.item = item.Value;

            if (this.timer != null)

            {

                this.timer.Change(-1, -1);

            }

            Complete(false, item.Exception);

        }

    }

  

    struct Item

    {

        T value;

        Exception exception;

        ItemDequeuedCallback dequeuedCallback;

  

        public Item(T value, ItemDequeuedCallback dequeuedCallback)

            : this(value, null, dequeuedCallback)

        {

        }

  

        public Item(Exception exception, ItemDequeuedCallback dequeuedCallback)

            : this(null, exception, dequeuedCallback)

        {

        }

  

        Item(T value, Exception exception, ItemDequeuedCallback dequeuedCallback)

        {

            this.value = value;

            this.exception = exception;

            this.dequeuedCallback = dequeuedCallback;

        }

  

        public Exception Exception

        {

            get { return this.exception; }

        }

  

        public T Value

        {

            get { return value; }

        }

  

        public ItemDequeuedCallback DequeuedCallback

        {

            get { return dequeuedCallback; }

        }

  

        public void Dispose()

        {

            if (value != null)

            {

                if (value is IDisposable)

                {

                    ((IDisposable)value).Dispose();

                }

                else if (value is ICommunicationObject)

                {

                    ((ICommunicationObject)value).Abort();

                }

            }

        }

  

        public T GetValue()

        {

            if (this.exception != null)

            {

                throw this.exception;

            }

  

            return this.value;

        }

    }

  

    class WaitQueueWaiter : IQueueWaiter

    {

        bool itemAvailable;

        ManualResetEvent waitEvent;

        object thisLock = new object();

  

        public WaitQueueWaiter()

        {

            waitEvent = new ManualResetEvent(false);

        }

  

        object ThisLock

        {

            get

            {

                return this.thisLock;

            }

        }

  

        public void Set(bool itemAvailable)

        {

            lock (ThisLock)

            {

                this.itemAvailable = itemAvailable;

                waitEvent.Set();

            }

        }

  

        public bool Wait(TimeSpan timeout)

        {

            if (timeout == TimeSpan.MaxValue)

            {

                waitEvent.WaitOne();

            }

            else if (!waitEvent.WaitOne(timeout, false))

            {

                return false;

            }

  

            return this.itemAvailable;

        }

    }

  

    class AsyncQueueWaiter : AsyncResult, IQueueWaiter

    {

        static TimerCallback timerCallback = new TimerCallback(AsyncQueueWaiter.TimerCallback);

        Timer timer;

        bool itemAvailable;

        object thisLock = new object();

  

        public AsyncQueueWaiter(TimeSpan timeout, AsyncCallback callback, object state)

            : base(callback, state)

        {

            if (timeout != TimeSpan.MaxValue)

            {

                this.timer = new Timer(timerCallback, this, timeout, TimeSpan.FromMilliseconds(-1));

            }

        }

  

        object ThisLock

        {

            get

            {

                return this.thisLock;

            }

        }

  

        public static bool End(IAsyncResult result)

        {

            AsyncQueueWaiter waiterResult = AsyncResult.End<AsyncQueueWaiter>(result);

            return waiterResult.itemAvailable;

        }

  

        static void TimerCallback(object state)

        {

            AsyncQueueWaiter thisPtr = (AsyncQueueWaiter)state;

            thisPtr.Complete(false);

        }

  

        public void Set(bool itemAvailable)

        {

            bool timely;

  

            lock (ThisLock)

            {

                timely = (this.timer == null) || this.timer.Change(-1, -1);

                this.itemAvailable = itemAvailable;

            }

  

            if (timely)

            {

                Complete(false);

            }

        }

    }

  

    class ItemQueue

    {

        Item[] items;

        int head;

        int pendingCount;

        int totalCount;

  

        public ItemQueue()

        {

            items = new Item[1];

        }

  

        public Item DequeueAvailableItem()

        {

            if (totalCount == pendingCount)

            {

                Debug.Assert(false, "ItemQueue does not contain any available items");

                throw new Exception("Internal Error");

            }

            return DequeueItemCore();

        }

  

        public Item DequeueAnyItem()

        {

            if (pendingCount == totalCount)

                pendingCount--;

            return DequeueItemCore();

        }

  

        void EnqueueItemCore(Item item)

        {

            if (totalCount == items.Length)

            {

                Item[] newItems = new Item[items.Length * 2];

                for (int i = 0; i < totalCount; i++)

                    newItems[i] = items[(head + i) % items.Length];

                head = 0;

                items = newItems;

            }

            int tail = (head + totalCount) % items.Length;

            items[tail] = item;

            totalCount++;

        }

  

        Item DequeueItemCore()

        {

            if (totalCount == 0)

            {

                Debug.Assert(false, "ItemQueue does not contain any items");

                throw new Exception("Internal Error");

            }

            Item item = items[head];

            items[head] = new Item();

            totalCount--;

            head = (head + 1) % items.Length;

            return item;

        }

  

        public void EnqueuePendingItem(Item item)

        {

            EnqueueItemCore(item);

            pendingCount++;

        }

  

        public void EnqueueAvailableItem(Item item)

        {

            EnqueueItemCore(item);

        }

  

        public void MakePendingItemAvailable()

        {

            if (pendingCount == 0)

            {

                Debug.Assert(false, "ItemQueue does not contain any pending items");

                throw new Exception("Internal Error");

            }

            pendingCount--;

        }

  

        public bool HasAvailableItem

        {

            get { return totalCount > pendingCount; }

        }

  

        public bool HasAnyItem

        {

            get { return totalCount > 0; }

        }

  

        public int ItemCount

        {

            get { return totalCount; }

        }

    }

}

And in order to cooperate with this InputQueue<T> we also need a asynchronous result class, which will be inherited from the IAsyncResult interface.

       1: /// <summary>

       2: /// A generic base class for IAsyncResult implementations

       3: /// that wraps a ManualResetEvent.

       4: /// </summary>

       5: abstract class AsyncResult : IAsyncResult

       6: {

       7:     AsyncCallback callback;

       8:     object state;

       9:     bool completedSynchronously;

    bool endCalled;

    Exception exception;

    bool isCompleted;

    ManualResetEvent manualResetEvent;

    object thisLock;

  

    protected AsyncResult(AsyncCallback callback, object state)

    {

        this.callback = callback;

        this.state = state;

        this.thisLock = new object();

    }

  

    public object AsyncState

    {

        get

        {

            return state;

        }

    }

  

    public WaitHandle AsyncWaitHandle

    {

        get

        {

            if (manualResetEvent != null)

            {

                return manualResetEvent;

            }

  

            lock (ThisLock)

            {

                if (manualResetEvent == null)

                {

                    manualResetEvent = new ManualResetEvent(isCompleted);

                }

            }

  

            return manualResetEvent;

        }

    }

  

    public bool CompletedSynchronously

    {

        get

        {

            return completedSynchronously;

        }

    }

  

    public bool IsCompleted

    {

        get

        {

            return isCompleted;

        }

    }

  

    object ThisLock

    {

        get

        {

            return this.thisLock;

        }

    }

  

    // Call this version of complete when your asynchronous operation is complete.  This will update the state

    // of the operation and notify the callback.

    protected void Complete(bool completedSynchronously)

    {

        if (isCompleted)

        {

            // It is a bug to call Complete twice.

            throw new InvalidOperationException("Cannot call Complete twice");

        }

  

        this.completedSynchronously = completedSynchronously;

  

        if (completedSynchronously)

        {

            // If we completedSynchronously, then there is no chance that the manualResetEvent was created so

            // we do not need to worry about a race condition.

            Debug.Assert(this.manualResetEvent == null, "No ManualResetEvent should be created for a synchronous AsyncResult.");

            this.isCompleted = true;

        }

        else

        {

            lock (ThisLock)

            {

                this.isCompleted = true;

                if (this.manualResetEvent != null)

                {

                    this.manualResetEvent.Set();

                }

            }

        }

  

        // If the callback throws, there is a bug in the callback implementation

        if (callback != null)

        {

            callback(this);

        }

    }

  

    // Call this version of complete if you raise an exception during processing.  In addition to notifying

    // the callback, it will capture the exception and store it to be thrown during AsyncResult.End.

    protected void Complete(bool completedSynchronously, Exception exception)

    {

        this.exception = exception;

        Complete(completedSynchronously);

    }

  

    // End should be called when the End function for the asynchronous operation is complete.  It

    // ensures the asynchronous operation is complete, and does some common validation.

    protected static TAsyncResult End<TAsyncResult>(IAsyncResult result)

        where TAsyncResult : AsyncResult

    {

        if (result == null)

        {

            throw new ArgumentNullException("result");

        }

  

        TAsyncResult asyncResult = result as TAsyncResult;

  

        if (asyncResult == null)

        {

            throw new ArgumentException("Invalid async result.", "result");

        }

  

        if (asyncResult.endCalled)

        {

            throw new InvalidOperationException("Async object already ended.");

        }

  

        asyncResult.endCalled = true;

  

        if (!asyncResult.isCompleted)

        {

            asyncResult.AsyncWaitHandle.WaitOne();

        }

  

        if (asyncResult.manualResetEvent != null)

        {

            asyncResult.manualResetEvent.Close();

        }

  

        if (asyncResult.exception != null)

        {

            throw asyncResult.exception;

        }

  

        return asyncResult;

    }

}

  

//An AsyncResult that completes as soon as it is instantiated.

class CompletedAsyncResult : AsyncResult

{

    public CompletedAsyncResult(AsyncCallback callback, object state)

        : base(callback, state)

    {

        Complete(true);

    }

  

    public static void End(IAsyncResult result)

    {

        AsyncResult.End<CompletedAsyncResult>(result);

    }

}

  

//A strongly typed AsyncResult

abstract class TypedAsyncResult<T> : AsyncResult

{

    T data;

  

    protected TypedAsyncResult(AsyncCallback callback, object state)

        : base(callback, state)

    {

    }

  

    public T Data

    {

        get { return data; }

    }

  

    protected void Complete(T data, bool completedSynchronously)

    {

        this.data = data;

        Complete(completedSynchronously);

    }

  

    public static T End(IAsyncResult result)

    {

        TypedAsyncResult<T> typedResult = AsyncResult.End<TypedAsyncResult<T>>(result);

        return typedResult.Data;

    }

}

  

//A strongly typed AsyncResult that completes as soon as it is instantiated.

class TypedCompletedAsyncResult<T> : TypedAsyncResult<T>

{

    public TypedCompletedAsyncResult(T data, AsyncCallback callback, object state)

        : base(callback, state)

    {

        Complete(data, true);

    }

  

    public new static T End(IAsyncResult result)

    {

        TypedCompletedAsyncResult<T> completedResult = result as TypedCompletedAsyncResult<T>;

        if (completedResult == null)

        {

            throw new ArgumentException("Invalid async result.", "result");

        }

  

        return TypedAsyncResult<T>.End(completedResult);

    }

}

I’m not going to explain more deeply into the InputQueue<T> and the AsyncResult classes. If you want to know more about it there’s a very good blog post to be reference.

Then with this new scaffold let’s amend our ChannelListener. First we will define a local variant of the InputQueue<T>, where the type parameter would be the IReplyChannel. This means we will queue all request channel in this class. We also need a local variant for currently using channel and an object for locking.

       1: private readonly InputQueue<IReplyChannel> _channelQueue;

       2: private readonly object _currentChannelLock;

       3: private IReplyChannel _currentChannel;

And initialize them in the constructor.

       1: public MessageBusReplyChannelListener(MessageBusTransportBindingElement transportElement, BindingContext context)

       2:     : base(context.Binding)

       3: {

       4:     _encoderFactory = (new TextMessageEncodingBindingElement()).CreateMessageEncoderFactory();

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

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

       7:     _bus = transportElement.Bus;

       8:  

       9:     _channelQueue = new InputQueue<IReplyChannel>();

      10:     _currentChannelLock = new object();

      11:     _currentChannel = null;

      12: }

Since the InputQueue<T> need to execute some procedures when terminate we have to implement the OnAbort, OnClose and OnClosed virtual methods to close the InputQueue<T>.

       1: protected override void OnAbort()

       2: {

       3:     try

       4:     {

       5:         lock (ThisLock)

       6:         {

       7:             _channelQueue.Close();

       8:         }

       9:     }

    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 { }

}

And then we will create an internal method that ensure the current channel is available. It will check if the local variant _currentChannel is null. If yes then it will create a new channel from another private method called CreateChannel, which we will implement in the next step, and assign the OnChannelClosed method to its Closed event, and then enqueued into the InputQueue<IRequestChannel>. If not, which means there’s a channel object available, we will not do anything here.

       1: private void EnsureChannelAvailable()

       2: {

       3:     IReplyChannel newChannel = null;

       4:     bool channelCreated = false;

       5:  

       6:     if ((newChannel = _currentChannel) == null)

       7:     {

       8:         lock (_currentChannelLock)

       9:         {

            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);

    }

}

  

private IReplyChannel CreateChannel(

    BufferManager bufferManager, MessageEncoderFactory encoder, EndpointAddress localAddress, MessageBusReplyChannelListener parent, IBus bus)

{

    return new MessageBusReplyChannel(_bufferManager, _encoder, localAddress, parent, _bus);

}

  

private void OnChannelClosed(object sender, EventArgs e)

{

    var channel = sender as IReplyChannel;

    lock (_currentChannelLock)

    {

        if (channel == _currentChannel)

        {

            _currentChannel = null;

        }

    }

}

Now we can use them to finish the asynchronous BeginOnAcceptChannel, EndOnAcceptChannel, and we will also amend the synchronous OnAcceptChannel as well.

       1: protected override IReplyChannel OnAcceptChannel(TimeSpan timeout)

       2: {

       3:     if (!IsDisposed)

       4:     {

       5:         EnsureChannelAvailable();

       6:     }

       7:  

       8:     IReplyChannel channel = null;

       9:     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 IReplyChannel OnEndAcceptChannel(IAsyncResult result)

{

    IReplyChannel channel;

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

    {

        return channel;

    }

    else

    {

        throw new TimeoutException();

    }

}

Server: ReplyChannel and RequestContext

The ReplyChannel created from the ChannelListener also need to support the asynchronous functions in order to make it work with the ServiceHost. In this case we will just use the asynchronous delegate invoke. Since we have had the synchronous ReceiveRequest method, what we need to do is to create a delegate from it, and invoke its Begin and End invoke method in the BeginTryRecevieRequest and the EndTryReceiveRequest methods.

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

       2: private TryReceiveRequestDelegate _tryReceiveRequestDelegate;

Initialize this delegate based on what we did in the ReceiveRequest method.

       1: public MessageBusReplyChannel(

       2:     BufferManager bufferManager, MessageEncoderFactory encoder, EndpointAddress address,

       3:     MessageBusReplyChannelListener parent,

       4:     IBus bus)

       5:     : base(parent)

       6: {

       7:     _bufferManager = bufferManager;

       8:     _encoder = encoder.CreateSessionEncoder();

       9:  

      10:     _localAddress = address;

      11:     _bus = bus;

      12:     _aLock = new object();

      13:  

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

      15:     {

      16:         rc = null;

      17:         // receive the request message from the bus

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

      19:         // box the wcf message

      20:         var raw = Encoding.UTF8.GetBytes(busMsg.Content);

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

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

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

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

      25:         // initialize the request context and return

      26:         rc = new MessageBusRequestContext(message, this, _bufferManager, _encoder, _localAddress, _bus, busMsg.MessageID);

      27:         return true;

      28:     };

      29: }

And invoke it asynchronously in the BeginTryRecevieRequest and the EndTryReceiveRequest methods.

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

       2: {

       3:     RequestContext context;

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

       5: }

       6:  

       7: public bool EndTryReceiveRequest(IAsyncResult result, out RequestContext context)

       8: {

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

      10:     return ret;

      11: }

The ReplyChannel had been modified finished and we don’t need to change anything in the RequestContext. Then let’s have a try.

Use the ServiceHost and ChannelFactory Mode

Now we will use the way that you are familiar with to test our transport extension. There is no channel listener, no channel in our test code, and the service business logic will not be mixed within the service hosting code. Let’s have a look.

First we will create a service contract interface and a service class which implement it. Again, we will use the string reverse as the sample.

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

       2: public interface ISampleService

       3: {

       4:     [OperationContract]

       5:     string Reverse(string content);

       6: }

       7:  

       8: public class SampleService : ISampleService

       9: {

      10:     public string Reverse(string content)

      11:     {

      12:         return new string(content.Reverse().ToArray());

      13:     }

      14: }

And in the main function we will create and open the service by using the ServiceHost. But since we didn’t implement any configuration extension so for now we have to use the code to set the binding and address.

       1: static void Main(string[] args)

       2: {

       3:     var bus = new InProcMessageBus();

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

       5:  

       6:     // establish the service

       7:     var host = new ServiceHost(typeof(SampleService));

       8:     var svcBinding = new MessageBusTransportBinding(bus);

       9:     host.AddServiceEndpoint(typeof(ISampleService), svcBinding, address);

      10:     host.Opened += (sender, e) =>

      11:         {

      12:             Console.WriteLine("Service opened at {0}", host.Description.Endpoints[0].ListenUri);

      13:         };

      14:     host.Open();

      15: }

And we will use the ChannelFactory<TChannel> to create a client side proxy.

       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:     ... ...

       8:  

       9:     // establish the client

      10:     var cliBinding = new MessageBusTransportBinding(bus);

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

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

      13:         {

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

      15:         };

      16:     var proxy = factory.CreateChannel();

      17: }

Then finally invoke the service by the proxy the ChannelFactory generate to us.

       1: static void Main(string[] args)

       2: {

       3:     var bus = new InProcMessageBus();

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

       5:  

       6:     // establish the service

       7:     ... ...

       8:  

       9:     // establish the client

      10:     ... ...

      11:  

      12:     // invoke the service

      13:     using (proxy as IDisposable)

      14:     {

      15:         Console.WriteLine("Say something...");

      16:         var content = Console.ReadLine();

      17:         while (!string.IsNullOrEmpty(content))

      18:         {

      19:             var result = proxy.Reverse(content);

      20:             Console.WriteLine("RESULT: {0} => {1}", content, result);

      21:             Console.WriteLine("Say something...");

      22:             content = Console.ReadLine();

      23:         }

      24:     }

      25: }

And this is the screenshot when I was running this code on my machine.

But don’t forget our goal, scaling-out on the server instances. Similar as what we did before, we will create more than one ServiceHost instance that listening on the same endpoint, which can handle the client request at the same time, to simulate the cross process or cross machine scaling-out scenario.

       1: class Program

       2: {

       3:     static void Main(string[] args)

       4:     {

       5:         var bus = new InProcMessageBus();

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

       7:  

       8:         // establish the services

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

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

        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("Say something...");

            var content = Console.ReadLine();

            while (!string.IsNullOrEmpty(content))

            {

                var result = proxy.Reverse(content);

                Console.WriteLine("RESULT: {0} => {1}", content, result);

                Console.WriteLine("Say something...");

                content = Console.ReadLine();

            }

        }

    }

  

    static ServiceHost EstablishServiceHost<TChannel, TService>(IBus bus, string address)

    {

        var host = new ServiceHost(typeof(TService));

        var binding = new MessageBusTransportBinding(bus);

        host.AddServiceEndpoint(typeof(TChannel), binding, address);

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

        {

            Console.WriteLine("Service ({0}) opened at {1}", host.GetHashCode(), host.Description.Endpoints[0].ListenUri);

        };

        host.Open();

        return host;

    }

}

As you can see I created three hosts and one client. In order to display which service host is handing the request I add one line code inside the service implementation class that display the current service host’s hash code.

       1: public class SampleService : ISampleService

       2: {

       3:     public string Reverse(string content)

       4:     {

       5:         Console.WriteLine("SERVICE: Invoked on the service host instance {0}.", OperationContext.Current.Host.GetHashCode());

       6:         return new string(content.Reverse().ToArray());

       7:     }

       8: }

So let’s have a look on what’s going on.

As you can see I invoked the service method four times and the first three of them was handled by the three different service host, which means scaling-out to my service instances. And the fourth one was handled by the first service host instance.

Summary

In this post I described how to amend our extension so that we can use it in a better way for service definition, implementation, hosting and client invoking. Once we implemented the relevant asynchronous methods in the ChannelListener and ReplyChannel we can use the ServiceContract interface to define the service contract, and use the ServiceHost to let WCF create the service class instance, open the service. On the client side we can use the ChannelFactory<TChannel> to get the proxy class of our ServiceContract interface. All of them are what we had been doing for a long time and familiar with.

Till now we had finished the basis of the transport extension, even though we had just implemented the request reply MEP. But as what I said in the first post, all other MEPs can be implemented by introducing the more ChannelListener, ChannelFactory and Channel.

But things always not as simple as we expected. In the next post I will show you how to implement the datagram MEP, with some code refactoring, with some bug fix.

You can download the source code of this post 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/cme

Posted On Friday, March 23, 2012 11:25 AM
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 3 – Service Host and Client Proxy Support

Client: ChannelFactory and RequestChannel

Server: ChannelListener

Server: ReplyChannel and RequestContext

Use the ServiceHost and ChannelFactory Mode

Summary

Posted On Friday, March 23, 2012 11:25 AM
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 3 – Service Host and Client Proxy Support

Client: ChannelFactory and RequestChannel

Server: ChannelListener

Server: ReplyChannel and RequestContext

Use the ServiceHost and ChannelFactory Mode

Summary

Posted On Friday, March 23, 2012 11:25 AM Filed Under [ Architecture Cloud Computing ]

Comments

Posted On Friday, March 23, 2012 11:25 AM
Filed Under [ Architecture Cloud Computing ]