Introduction
I work in an environment that utilizes BizTalk to process messages from various sources and distribute to numerous send ports. Now, one could utilize BizTalk as an ESB; however, this poses some issues, mainly a new send port needs to be created for each new application which needs messages routed to it. That is, unless one could guarantee all developers would only peek into a single message queue and never dequeue a message. Honestly, that would lead to more problems than it solves, not to mention it would be terrible practice. This led me to using WCF Callback Services as a single entry point in our ESB (Enterprise Service Bus).
Background
Considering WCF (Windows Communication Foundation) supports callback contracts and thus Event Driven Architecture, one could easily create a send port in BizTalk to a message queue (MSMQ, in this case), then expose a WCF Event Driven Web Service which numerous clients could easily subscribe to and receive messages from. Along my journey, I came across a lot of questions and issues. This is the reason I’m writing this entry today, to hopefully stop some of you from wasting too much time looking.
Using the Code
Let’s start with the easy part, the service contracts. This includes the callback contract as well as operations to allow clients to subscribe and unsubscribe. I will use the most basic web service I can think of, as to not stray from the actual issues I came across.
Collapse | Copy Code
[ServiceContract(SessionMode=SessionMode.Required,
CallbackContract=typeof(IEventSystemCallback))]
interface IEventSystem
{
[OperationContract(IsOneWay=true)]
void Subscribe();
[OperationContract(IsOneWay = true)]
void Unsubscribe();
}
interface IEventSystemCallback
{
[OperationContract(IsOneWay = true)]
void OnMessageReceived(string message);
}
Now, this is really all it takes to generate an event driven web service in WCF; at least, as far as the contracts are concerned. If you notice, there are two operations which will be exposed through the service (Subscribe andUnsubscribe). Remember, this is stripped to its basics for demonstration purposes. In the actual ESB solution, I have an object which represents the actual events one would like to subscribe to as well as other details for tracking purposes.
The IEventSystemCallback interface is the callback contract, as you will notice in the Service Contract for theIEventSystem interface. Therefore, you do not need a service contract attribute on the IEventSystemCallbackcontract. One must only expose all of the methods (or events) the client should expect.
Now, let’s get into the hard stuff. We’ll start with the service itself. I’ll show you the entire service implementation, and then we’ll break it down.
Collapse | Copy Code
[ServiceBehavior(InstanceContextMode=InstanceContextMode.PerSession)]
internal sealed class EventService:IEventSystem
{
public delegate void CallbackDelegate<T>(T t);
public static CallbackDelegate<string> MessageReceived;
#region IEventSystem Members
public void Subscribe()
{
IEventSystemCallback callback =
OperationContext.Current.GetCallbackChannel<IEventSystemCallback>();
MessageReceived += callback.OnMessageReceived;
ICommunicationObject obj = (ICommunicationObject)callback;
obj.Closed += new EventHandler(EventService_Closed);
obj.Closing += new EventHandler(EventService_Closing);
}
void EventService_Closing(object sender, EventArgs e)
{
Console.WriteLine("Client Closing...");
}
void EventService_Closed(object sender, EventArgs e)
{
MessageReceived -= ((IEventSystemCallback)sender).OnMessageReceived;
Console.WriteLine("Closed Client Removed!");
}
public void Unsubscribe()
{
IEventSystemCallback callback =
OperationContext.Current.GetCallbackChannel<IEventSystemCallback>();
MessageReceived -= callback.OnMessageReceived;
}
#endregion
public static void SendMessage(string Message)
{
try
{
MessageReceived(Message);
}
catch (Exception ex)
{
Console.WriteLine(ex.ToString());
}
}
public static void NotifyServiceStop()
{
SendMessage("Service Stopping...");
}
}
First, let’s start with the service itself:
Collapse | Copy Code
[ServiceBehavior(InstanceContextMode=InstanceContextMode.PerSession)]
internal sealed class EventService:IEventSystem
Here, we have implemented the IEventSystem interface and added a service behavior attribute to set theInstanceContextMode. You can set the InstanceContextMode to PerCall, PerSession, or Single. We will go into this more in the Throttling section, later on. For now, just know I chose PerSession because the use of callback contracts requires a session, and I just wanted to make it easier for explaining throttling later.
*Note: InstanceContextMode.PerSession and SessionMode=SessionMode.Required have no relation. You could have any InstanceContextMode, with Required SessionMode, as we will see later.
Collapse | Copy Code
public delegate void CallbackDelegate<T>(T t);
public static CallbackDelegate<string> MessageReceived;
One of the most difficult things to grasp about a Callback Service is where to store all of the subscribers. This is why I’ve created the static GenericDelegate. Normally, I would place all my delegates inside another namespace and file location, but for ease of demo, I put the delegate creation right here in the service class. You could also store your subscriber reference (or InstanceContexts, if you will) inside a static collection object (List<T>, ArrayList,Hashtable, etc.). I chose a delegate because it’s really easy to utilize, and it is also consistent with how an Event Service works. All of the Instance Contexts will be stored as the target of the InvocationList object when you wire up the events, as we will see here.
Collapse | Copy Code
public void Subscribe()
{
IEventSystemCallback callback =
OperationContext.Current.GetCallbackChannel<IEventSystemCallback>();
MessageReceived += callback.OnMessageReceived;
ICommunicationObject obj = (ICommunicationObject)callback;
obj.Closed += new EventHandler(EventService_Closed);
obj.Closing += new EventHandler(EventService_Closing);
}
void EventService_Closing(object sender, EventArgs e)
{
Console.WriteLine("Client Closing...");
}
void EventService_Closed(object sender, EventArgs e)
{
MessageReceived -= ((IEventSystemCallback)sender).OnMessageReceived;
Console.WriteLine("Closed Client Removed!");
}
public void Unsubscribe()
{
IEventSystemCallback callback =
OperationContext.Current.GetCallbackChannel<IEventSystemCallback>();
MessageReceived -= callback.OnMessageReceived;
}
The ClientChannel is passed by the subscriber automatically with every service method call. So, when they subscribe, we proceed to wire up all of the events utilized in this demo. It is really only necessary to wire up theOnMessageReceived method of the callback contract, but how do you handle disconnected clients? I’ve seen it done where the service just tries to monitor the channel state with every callback call, but why go through all the overhead when you can just wire up the Closing and /or Closed events of the ICommunicationObject (or IDuplexContextChannel, if you choose this cast)? There are also other useful events you can wire up on the channel, such as: Faulted, Opening, and Opened.
Lastly, I create a few methods to actually send the messages to the subscribers. With the actual ESB Service, This is where the MSMQ monitoring would occur and fire all the new messages to the subscribers.
Collapse | Copy Code
public static void SendMessage(string Message)
{
try
{
MessageReceived(Message);
}
catch (Exception ex)
{
Console.WriteLine(ex.ToString());
}
}
public static void NotifyServiceStop()
{
SendMessage("Service Stopped!");
}
I also added a NotifyServiceStop method for the same reason that I wired up the subscriber's Closing andClosed events. It’s only nice to notify the subscribers that the service is going down. In my implementation, I have aServiceNotification object to allow for multiple types of notifications to be sent, the same way I have theEventNotifications for multiple events to be transmitted.
I have hosted this demo in a console application using a ServiceHost. You can self host the same way using a Windows service, or utilize WAS in IIS 7 to host the netTcp Service. There are also a few other ways to accomplish this, but that’s outside the scope of this article.
Collapse | Copy Code
class Program
{
private static int count;
public static int Count
{
get
{
count += 1;
return count;
}
}
static void Main(string[] args)
{
ServiceHost host = new ServiceHost(typeof(EventService));
host.Closing += new EventHandler(host_Closing);
host.Open();
Console.WriteLine("Service Started");
Timer timer = new Timer(delegate(object state)
{
string message = "Message " + Count.ToString();
try
{
if (EventService.MessageReceived != null)
{
Console.WriteLine(
EventService.MessageReceived.GetInvocationList().Length.ToString() +
" Subscribers");
EventService.SendMessage(message);
Console.WriteLine("Sent: " + message);
}
else
{
Console.WriteLine("0 Subscribers");
Console.WriteLine("Skipped: " + message);
}
}
catch (Exception ex)
{
Console.WriteLine(ex.ToString());
}
}, null, 5000, 5000);
Console.ReadKey(true);
host.Close();
}
static void host_Closing(object sender, EventArgs e)
{
try
{
EventService.NotifyServiceStop();
}
catch (Exception ex)
{
Console.WriteLine(ex.ToString());
}
}
}
There’s really no magic here. Just your ordinary ServiceHost implementation. The only things to point out here are that I wired up the host.Closing event to notify the subscribers of the service closing, and that with the use of the static delegate for storing the subscribers, you can inspect different aspects of the InvocationList. Here, I just show how many subscribers are listening.
Onto the configuration …
This is where most of the confusion came for me, and in fact, there are still some questions I have that you might be able to help out answering.
Collapse | Copy Code
="1.0" ="utf-8"
<configuration>
<system.serviceModel>
<behaviors>
<serviceBehaviors>
<behavior name="ExposeMexAndThrottleBehavior">
<serviceMetadata httpGetEnabled="true"
httpGetUrl="http://localhost:1111/EventService/Mex"/>
<serviceThrottling maxConcurrentCalls="3"
maxConcurrentInstances="100"
maxConcurrentSessions="100"/>
</behavior>
</serviceBehaviors>
</behaviors>
<services>
<service name="ESBService.EventService"
behaviorConfiguration="ExposeMexAndThrottleBehavior">
<endpoint address="net.tcp://localhost:9999/EventService/"
binding="netTcpBinding"
contract="ESBService.IEventSystem"/>
<endpoint address="http://localhost:1111/EventService/Mex"
binding="mexHttpBinding"
contract="IMetadataExchange"/>
</service>
</services>
</system.serviceModel>
</configuration>
I won’t be going into how to create an endpoint or expose metadata, as there are numerous examples on the web. The interesting thing to look at in this config is the serviceThrottling. When working with my first WCF duplex services, this was the most confusing. In fact, I didn’t even think I needed it because I didn’t want to “Throttle” my service at all. I would start up my service and start connecting clients one after the other, but I would continue to get stuck at 10 maximum clients subscribed. I tried setting the MaxConnections property on the netTcpBinding, with no luck.
Collapse | Copy Code
<bindings>
<netTcpBinding>
<binding name="tcpSettings" maxConnections="50"/>
</netTcpBinding>
</bindings>
This setting is still confusing to me actually, because no matter what I set it to, it has no effect on my service. I read that you can “Throttle” the service using either the serviceThrottling or the maxConnections property, and the service will utilize the lowest setting between the two. Doesn’t seem to make any difference at all what I set theMaxConnections to, so don’t even use it.
The solution was presented with the use of the serviceThrottling setting, which is actually misleading in my opinion. The word throttling to me means limiting the Instances, Sessions, and Calls, when actually they are already “Throttled”, by default. If you take out the serviceThrottling behavior and inspect theServiceHost.Description.Behaviors while the service is running, you will not see aServiceThrottlingBehavior. This led me to believe that serviceThrottling was not implemented, unless specified explicitly. Therefore, I kept going down the path that it must have something to do with my TcpBindingsettings. Come to find out that even though you can’t see the throttling behavior on the service, it’s there, throttling your service to unrealistic limits. Perhaps, a better name for the serviceThrottling behavior could beserviceLimiting.
To describe what these throttling settings mean, let’s go into them now.
Remember that I used a PerSession InstanceContextMode. That affects how I set my throttling parameters.
Collapse | Copy Code
<serviceThrottling maxConcurrentCalls="3"
maxConcurrentInstances="100"
maxConcurrentSessions="100"/>
The maxConcurrentSessions is the most important here. This allows 100 subscribers to connect to my service, in this example. With each subscription, the service will create a new session, and therefore a new instance, thusmaxConcurrentInstances is set the same. In the real world, you will set these settings based upon your hardware constraints and consumers expected plus room to grow.
The maxConcurrentCalls should be set between 1 and 3 percent of your maxConcurrentInstances. This is why mine is set to 3.
Default values when not specified (or no serviceThrottling behavior is created):
maxConcurrentCalls = 16maxConcurrentSessions = 10maxConcurrentInstances = Unlimited
Now, your throttling will be set differently if you use a PerCall or Single InstanceContextMode. I won’t go into each configuration, in this demo. If you need help with them, please feel free to post, and I'll try to get back to you quickly.
Client code:
Collapse | Copy Code
public partial class Form1 : Form,ESB.IEventSystemCallback
{
public Form1()
{
InitializeComponent();
}
ESB.EventSystemClient client;
private void Form1_Load(object sender, EventArgs e)
{
try
{
client =
new ESBClient.ESB.EventSystemClient(new InstanceContext(this));
client.Subscribe();
}
catch (Exception ex)
{
listBox1.Items.Add(ex.ToString());
}
}
#region IEventSystemCallback Members
public void OnMessageReceived(string message)
{
listBox1.Items.Add(message);
}
#endregion
* To use the sample code, open the client and service solutions in two separate VS2008 IDEs. Run the Service, then use CTRL+F5 to open multiple clients. Open and close as many clients you want. You can
