Event-Driven Communication Foundation

• Download demo Chat Room project - 21.2 KB
• Download source - 13.4 KB

Introduction

It is common knowledge that WCF is a very powerful technology. Unfortunately, it is practically impossible to achieve this power without complex concepts such as proxies, asynchronous delegates, duplex channels and callback handlers. Even if you are an expert in these concepts, you will have to write tons of code to implement effective and reliable messaging.

Moreover, since GUI and Event-Driven Architecture have become standard, you will need to design Asynchronous Event Engines with multi-casting and filtering. I am sure these words do not scare you and you are definitely up to this work, but why not to take advantage of ready-to-use solutions that save you time and allow thinking more about your business tasks? Let's take a smooth look at one such solution.

I want to introduce a tool that I am going to use in all my projects. In a few words, this is an event-driven framework for WCF. We called it Event Driven Communication Foundation (EdCF) and it is based on the Publisher/Subscriber pattern. Puzzling code has been hidden and now all you need to do for organizing bidirectional, asynchronous, reliable messaging is just list your events and write event handlers.

Now I will try to partition our framework into functional pieces and get down to the nitty-gritty.

Publisher/Subscriber Pattern

First of all, I want to say several words about the Publisher/Subscriber pattern. Why have we chosen this pattern as the basis for the framework? Because there is a number of different communication scenarios: one-to-one, one-to-many, many-to-many. As soon as we decided that our framework should be able to implement all these scenarios, the most suitable design pattern was the Publisher/Subscriber one with an intermediate pub/sub service.

How does this work? Any remote peer takes the role of the publisher or/and subscriber. The publisher informs the intermediate Pub/Sub service that it is able to fire events. If a subscriber wants to get notification about any specific event, it asks the Pub/Sub service to add him to the list of interested subscribers. Whenever the publisher changes state, it fires an event. The intermediate Pub/Sub service stores a list of subscribers and multi-casts the event to all subscribers from this list. For more information, please refer to this article.

Asynchronous Messaging

This is the first tricky lock you will face if you've decided to do all the work by yourself. The scenario is simple: your server needs to process the same request from two different clients.

The first client has a slow connection. If your server knows nothing about asynchronous messaging, the second client will wait until the first request is processed. What should we do if we have dozens of concurrent clients? We should perform each request in the separated thread from a thread pool. This requires decisions on an architecture level and knowledge about asynchronous delegates. To illustrate how confusing this code may be, I show you the following:

I have the list of subscribers and want to publish the MyEvent event. I will write something like this:

public static Publish (TSub[] subscribers, string eventName)
{
    foreach (TSub subscriber in subscribers)
    {
        Invoke(subscriber, eventName)
    }
}

If I decided to do the same but asynchronously, I would have much more typing:

public static Publish (TSub[] subscribers, string eventName)
{
    WaitCallback callbackDelegate = delegate (TSub subscriber) 
    {
        Invoke(subscriber, eventName);
    }

    Action<TSub>queueUp = delegate(TSub subscriber)
    {
        ThreadPool.QueueUserWorkItem(callbackDelegate, subscriber)
    }

    Array.ForEach(subscriber, queueUp);
}

Fortunately, this entire outfit is hidden deep in the framework. From the end-user's point of view, it looks like a single method invocation:

PublishingService.MyEvent();

Result Collector

One of the advantages of the Publisher/Subscriber pattern with the intermediate Pub/Sub service is decoupling publishers and subscribers. The subscriber has no knowledge about the publisher and vice-versa. This is very important behavior, since you need to build a stable and extensible solution with hundreds of clients.

However, what if you need to implement communication between a few closely tied peers? In this case, the ability to get the result of event raising is disabled. For example, imagine the classic calculator scenario. A client connects to the server and asks it to sum two values and return the result. As you remember, we wanted to have a framework working in a wide spectrum of scenarios. Thus, the described problem should not bother our users. We should offer a solution and be able to pass results of the event to the publisher.

For example, any single subscriber should return a string. The publisher wants to have a list of strings from all subscribers as the result of the event raising. From the end-user standpoint, the most useful would be the following:

string [] myResult = PublishingService.MyEvent();

However, this code will freeze up the publisher thread until all subscribers reply. We are not going to have such a drawback. We want to collect results asynchronously and get them as soon as they are ready. The solution is to use callbacks. The publisher is able to raise any event described in the IMyEvents_Pub interface.

[ServiceContract]
public interface IMyEvents_Pub
{
    [OperationContract(IsOneWay = true)]
    [EventDrivenOperationContract(IsWithReply = true)]
    string MyEvent();
}

If you need to get the results of the event, you need to mark the corresponding method of the IMyEvent_Pub interface with the [EventDrivenOperationContract(IsWithReply = true)] attribute. Based on this information, the framework will automatically generate the whole infrastructure, interfaces, callback contracts, proxies and bindings. The only thing you need to do is just define a class implementing the IMyEvents_PubCallback interface on the publisher side. For more information, please refer to the "Code Generation" section of this document.

public class MyPublishCallbackHandler : IMyEvents_PubCallback
{
    public void MyEventReply(string[] returnedResults)
    {
        Console.WriteLine("Returned results: ");
        foreach (string s in returnedResults)
            Console.WriteLine("\t" + s);
    }
}

The publisher raises an event. As soon as the results are ready, the Pub/Sub service makes a callback to the publisher with an array of results as parameters.

Subscriber Filtering

The filtering of subscribers is quite a common task for distributed applications. It is set when the list of subscribers is not permanent and depends on the event we need to process. For example, take a simple Chat sample with two ChatRooms. If a subscriber enters the first ChatRoom, it is not interested in messages published in the second one. So, we should be able to prevent receipt of wrong messages. To take advantage of the filtering options of the framework, an end-user should do the following:

1. Define a Filter enum as a member of the Pub/Sub service:
   C#

   [Flags]
   public enum Filter
   {
       ChatRoom1 = 0x1,
       ChatRoom2 = 0x2
   }

2. Mark appropriate events with the [EventDrivenOperationContract(IsWithFilter = true)] attribute:
   C#

   [ServiceContract]
   public interface IMyEvents_Pub
   {
       [OperationContract(IsOneWay = true)]
       [EventDrivenOperationContract(IsWithFilter = true)]
       void MyEvent(Filter filter);
   }

3. Subscribe subscribers with the corresponding filter:
   C#

   SubscribeWithFilter("MyEvent", (int)Filter.ChatRoom1);

To subscribe with several filters, just use the bitwise OR operator:

SubscribeWithFilter("MyEvent", 
    (int)(Filter.ChatRoom1 | Filter.ChatRoom2));

Code Generation

This simplicity appears self-evident, but the framework is doing hard work behind the scenes. As I mentioned before, the framework is actively used in our everyday work. Sometimes it is very tedious to write the same pieces of code again and again. So, we decided to apply automatic code generation to eradicate some boring work. Although this code generation is widely used, the end-user will never face it. Thus, to save your and my time, I will not uncover all the places and dig all the code. I'll just show you an example and you will understand the gist.

Let's return to the "Result Collector" section. To use this functionality, the end-user should write only five short lines of code:

//Define an event and mark it with special atribute 
[EventDrivenOperationContract(IsWithReply = true)]
string MyEvent();

//Write a callback handler 
public class MyPublishCallbackHandler : IMyEvents_PubCallback
{
public void MyEventReply(string[] returnedResults)
{            
}
}

Very simple, isn't it? That's because the framework is doing all the work instead of us.

1. It generates the IPublishCallback interface. Publisher should realize this interface to collect the results. This interface contains methods with the Reply postfix corresponding to every event marked with the [EventDrivenOperationContract(IsWithReply = true)] attribute.
2. It offers the PublishCallbackBiulder builder, which generates instances of transparent proxy of type IPublishCallback. The proxy is retrieved from the OperationContext of the calls and actively use in the framework.
3. It wraps the IMyEvent_Pub interface, as it should require IPublishCallback implementation on the publisher side.
4. It generates the ReturnedResultsConverter class. As you understand, any single subscriber returns a single value, but a publisher should have a list of all results from all subscribers. The ReturnResultConverter class greatly helps here. It makes it possible to convert subscribers returned with results of type Object[] (System.Reflection.MethodInfo.Invoke() returns Object) to the array collection of the correct return type (e.g. if subscriber returns String then the publisher expects String[]).
5. The ReturnedResultsConverter class content corresponds to each event method marked with the ReplyResultsConverter postfix

Now take a look at the hidden code. Let's imagine we have a single MyEvent event, as defined in the beginning of the section.

1. The generated IPublishCallback looks like this:
   C#

   public interface IPublishCallback 
   {
       [OperationContract(IsOneWay=true)]
       void MyEventReply(string[] returnedResults);        
   }

2. The generated PublishCallBackBuilder looks like this:
   C#

   public class PublishCallbackBiulder 
   { 
       public static IPublishCallback publishCallback;
          
       public static void BuildCallback() 
       {
           publishCallback = OperationContext.Current.GetCallbackChannel<ipublishcallback />();
       }
   }

3. The generated proxy looks like this:
   C#

   [System.ServiceModel.ServiceContract(CallbackContract=
       typeof(EDSOAFW.IPublishCallback), Name="IMyEvents_Pub")]
       public interface IMyEvents_Pub_Wrap : EDSOA.IMyEvents_Pub {}

4. The generated ReturnedResultsConverter looks like this:
   C#

   public class ReturnedResultsConverter 
   {
       public static string[] MyEventReplyResultsConverter(
           object[] returnedResults) 
       {
           string[] returnArray = new string[returnedResults.Length];
           for (int i = 0; (i < returnedResults.Length); i = (i + 1))
           {
               returnArray[i] = (String)returnedResults[i];
           }
           return returnArray;
       }
   }

Of course, we can do this manually, but imagine that you have several decades of different events. You will have to continuously retype this code for each one. The framework will not allow you to die from boredom during tedious coding.

And the last accord: I want to demonstrate all this stuff with the help of a simple chat sample.

Chat Sample Introduction

First of all, let's define basic terms. This will guarantee that we are speaking the same language. The following terms will be widely used in the future, so please glance at the list to check whether you have the same meaning in your head.

Publisher/Subscriber Service is an intermediate remote peer, handling the list of publishers and subscribers. It conveys messages from a publisher to subscribers. This service is also responsible for message routing, batching and queuing. If we are talking in terms of client-server architecture, this is our server.

Publisher is the remote side initializing message sending. In a simple word, this is a client side that is able to send messages to the server.

Subscriber is the remote side waiting for a message. This is a client side that is able to receive messages.

Publisher/Subscriber Client is an application combining publisher and subscriber functionality.

Event is the process satisfying the following statements:

1. A publisher sends the same message to several subscribers.
2. All subscribers are called concurrently. This means that if the connection to a specific subscriber is slow (or is broken), sending to the rest of the clients will not be delayed until the problem subscriber answers.

Our sample implements simple chat. Thus, from the business point of view, we have a little bit different terminology. We have ChatRoom, which is nothing but Publisher/Subscriber service. We also have ChatClient, which is a client application that is able act as publisher and subscriber simultaneously.

There is a thing I want to mention before we go further. In my writing, I rely on the fact that you are familiar with basic WCF concepts. If you are not, of course you can continue with EdCF, but I would recommend finding some information about WCF basics. You can start with MSDN.

OK, we have agreed about the basic terms. Now we have a number of steps needed to build a chat sample. I put the description corresponding to each step in a separate HOWTO.

HOWTO: Create and Configure a Publisher/Subscriber Service (ChatRoom)

Currently, the sample has the only event: OnMessageSent. If you want to have additional events, you should just add corresponding methods to the IMyEvents_Pub and IMyEvents_Sub interfaces. Let's imagine you want to add a new event called MyNewEvent. The only changes you should do are just modifying the service contract interface for publishing...

[erviceContract]
public interface IMyEvents_Pub
{
    [OperationContract(IsOneWay = true)]
    [EventDrivenOperationContract(IsWithReply = true)]
    void OnMessageSent(string message);

    [OperationContract(IsOneWay = true)]
    void MyNewEvent();
}

...and the corresponding service contract for subscribing:

[erviceContract]
public interface IMyEvents_Sub
{
    [OperationContract(IsOneWay = false)]
    string OnMessageSent(string message);

    [OperationContract(IsOneWay = true)]
    void MyNewEvent();
}

The last thing you need to do is to change the BaseAddress value in the App.config file to the appropriate IP address.

<service name="ChatRoom.MySubscriptionService" 
    behaviorConfiguration="ForDebuggingBehavior">
<host>
    <baseAddresses>
        <add baseAddress="http://localhost:8000/MySubscriptionService"/>
    </baseAddresses>
</host>
…
<service name="ChatRoom.MyPublishService_Wrapper" 
    behaviorConfiguration="ForDebuggingBehavior">
<host>
    <baseAddresses>
        <add baseAddress="http://localhost:8000/MyPublishService"/>
    </baseAddresses>
</host>

HOHOWTO: Create a Proxy for Publishing/Subscribing

Well, we have configured and hosted WCF Service. Now we want to have a Proxy that allows our clients to connect to the service. As MSDN says, you have three options:

1. Retrieve the WSDL from the service and handcraft the proxy.
2. Use the Add Service Reference feature of Visual Studio 2005.
3. Use the SvcUtil.exe tool to generate proxy classes.

However, I would advise selecting another one. The ChatSample solution has already created proxies for publishing and subscribing. If you want to add a new event in your solution, just use these proxies with a small amount of modification. For example, say you want to add a new MyNewEvent.

ChatClient >> clientForSubscription.cs

public interface IMySubscriptionServiceCallback
{
    [System.ServiceModel.OperationContractAttribute(
        Action="http://tempuri.org/IMySubscriptionService/OnMessageSent", 
        ReplyAction=
        "http://tempuri.org/IMySubscriptionService/OnMessageSentResponse")]
        string OnMessageSent(string message);

    [System.ServiceModel.OperationContractAttribute(
        Action="http://tempuri.org/IMySubscriptionService/MyNewEvent",
        ReplyAction=
        "http://tempuri.org/IMySubscriptionService/MyNewEventResponse")]
    void MyNewEvent();
}

ChatClient >> clientForPublishing.cs

public interface IMyEvents_Pub
{    
    [System.ServiceModel.OperationContractAttribute(IsOneWay=true,
        Action="http://tempuri.org/IMyEvents_Pub/OnMessageSent")]
    void OnMessageSent(string message);

    [System.ServiceModel.OperationContractAttribute(IsOneWay=true,
        Action="http://tempuri.org/IMyEvents_Pub/MyNewEvent")]
    void MyNewEvent();
}

HOWTO: Configure a Publisher/Subscriber Client (ChatClient)

Since you prepared a Proxy, your client is able to connect to the service. However, you need to set up transport settings and addresses of services. All configurable values are located in the App.config file. Here you are able to change a number of parameters. This number is really huge and I would advise you to refer to the official documentation if you want to fine-tune your transport. However, if you want to start immediately, you need only to set up the service's address and the client base address for callbacks.

<client>
    <endpoint address="http://localhost:8000/MySubscriptionService"
        binding="wsDualHttpBinding" bindingConfiguration="sub1Binding"
        contract="IMySubscriptionService" name="MainEndpoint" />

    <endpoint address="http://localhost:8000/MyPublishService" 
        binding="wsDualHttpBinding"
        bindingConfiguration="pub1Binding" contract="IMyEvents_Pub" /> 
</client>

<bindings>
    <wsDualHttpBinding>
        <binding name="sub1Binding" openTimeout="00:10:00" 
            receiveTimeout="00:10:00" sendTimeout="00:10:00" 
            bypassProxyOnLocal="false" 
            clientBaseAddress="http://localhost:8000/myClient1sub/"
            useDefaultWebProxy="true" />

        <binding name="pub1Binding" openTimeout="00:10:00" 
            receiveTimeout="00:10:00" sendTimeout="00:10:00" 
            bypassProxyOnLocal="false" 
            clientBaseAddress="http://localhost:8000/myClient1pub/"
            useDefaultWebProxy="true" />

    </wsDualHttpBinding>
</bindings>

HOWTO: Subscribe to an Event and Publish an Event

We are close to the end. The whole infrastructure is ready and now we have to perform four last steps. Take a look at the rest of the code we should write.

1. The first step is to create a proxy for event subscription.
   C#

   MySubscriptionServiceClient subscriptionServiceClient = 
       new MySubscriptionServiceClient(
       subInstanceContext, endpointConfigurationName);

2. The second step is to subscribe to our event.
   C#

   subscriptionServiceClient.Subscribe("MyNewEvent");

3. The third step is to create a proxy for event publishing.
   C#

   MyEvents_PubClient eventsClient = 
       new MyEvents_PubClient(pubInstanceContext);

4. The last step is to publish the event.
   C#

   eventsClient.MyNewEvent();

That's all; you now have asynchronous, reliable messaging with multicast support.

Points of Interest

In conclusion of the article, I want to ask the CodeProject community to not hesitate in writing any comments and questions. I know many things about SOA and Event-Driven Architecture, but I want to know more. I am able to share my knowledge and I believe that you also have something to share. Please leave me a message in the discussion area at the bottom of this article.

History

• 21 September, 2007 -- Original version posted