Download the source code for this article.
I have spent a few days trying to find a solution for this problem. What I wanted to do was to have a client and a server talking to each other asynchronously using MSMQ as transport. For that effect, the service contracts used in this service have to be one-way only. First thing I had to do then was to create a service contract for the server and another one for the client that I was going to use for callbacks.
Please note that all code presented in this article was written in Visual Studio “Orcas” Beta 1 and therefore uses constructs that are exclusive to C# 3.0 and Linq. You will need this version of Visual Studio to execute the sample code. You will also need MSMQ installed on your machine and you need to create the two private transactional queues named .\private$\clientqueue and .\private$\serverqueue.
I created a single project that would have both service contracts and this project would be referenced by both server and client applications. The contracts look like the following:
[ServiceContract(
Namespace="http://schemas.example.com/servicecontracts/07/06/iservicecontract")]
public interface IServiceContract
{
[OperationContract(IsOneWay = true)]
void DoSomething(string somethingToDo);
}
[ServiceContract(
Namespace = "http://schemas.example.com/servicecontracts/07/06/icallbackcontract")]
public interface ICallbackContract
{
[OperationContract(IsOneWay = true)]
void DidSomething(string somethingDone);
}
The IServiceContract contract will be implemented by the server whereas the ICallbackContract will be implemented by the client. The implementation for the server looks like the following:
public class ServiceImplementation : IServiceContract
{
private static ChannelFactory<ICallbackContract> _channelFactory = InitializeFactory();
public void DoSomething(string somethingToDo)
{
string result = string.Format("Done ‘{0}’.", somethingToDo);
ICallbackContract client = _channelFactory.CreateChannel(new EndpointAddress("net.msmq://localhost/private/clientqueue"));
client.DidSomething(result);
}
}
As you can see the code is pretty simple. I just concatenate the input string to something else and call back the client. Notice though that I use a ChannelFactory<TChannel> instead of the traditional svcutil.exe generated client. This approach has the advantage of being quicker (as the factory is pre-initialised) and is necessary in this sample as I need to reference both client and server interfaces on both sides.
I then implemented some startup code in a console application. The code just launches an instance of ServiceHost to host the service implementation.
For the client, I have implemented the ICallbackContract interface in a class that look as follows:
public class CallbackImplementation : ICallbackContract
{
public void DidSomething(string somethingDone)
{
Console.WriteLine("Server says: {0}.", somethingDone);
}
}
The client console app then looks like the following:
class Program
{
private static ChannelFactory<IServiceContract> _channelFactory = new ChannelFactory<IServiceContract>("*");
static void
{
using (ServiceHost serviceHost = new ServiceHost(typeof(CallbackImplementation)))
{
serviceHost.Open();
Console.WriteLine("The client is online!");
string input = string.Empty;
do
{
Console.WriteLine("Tell the server what to do or type 'x' to exit:");
input = Console.ReadLine();
if (input != "x")
{
ThreadStart threadStart = delegate
{
IServiceContract client = _channelFactory.CreateChannel();
client.DoSomething(input);
};
var thread = new Thread(threadStart);
thread.IsBackground = true;
thread.Start();
}
} while (input != "x");
}
}
}
As you can see, the client program starts its ServiceHost instance and then allows the user to input strings that it will send to the server. The server will do something with the string and call back the client passing the results. The service implementation in the client will print the returned string to the console. Provided that you have configured the service and client appropriately (you can see the config files in the downloadable sample) you should be able to run this code and it should all work.
What I wanted to do, though, was to have the thread that has called the server service in the first place to wait for the server to reply and then get access to the data sent by the server. It sounds simple, but if you think that the thread that receives messages from the server doesn’t actually exists until the server sends a message, then it becomes quite tricky. I have then done a mixed salad using constructs in System.Threading and WCF to try to solve the problem and this solution I will present step-by-step below.
The first thing I have realised is that I didn’t want to change method signatures or pass parameters in and out of methods to achieve correlation. I wanted instead to have a header in the message that could be set by the client and read by the server and vice-versa. The value I was going to pass in this header is a Guid, as it is virtually unique. To inject headers in the message WCF allows you to create an implementation of System.ServiceModel.Dispatcher.IClientMessageInspector. When added to a WCF runtime, that class will then have the opportunity to inspect any messages before they get sent out to the server and after they get received by the client (responses). As my service is one-way only, I didn’t care about inspecting the response message. What I wanted was to inspect the output message and add my headers. I have also implemented the System.ServiceModel.Description.IEndpointBehavior interface in the same class, as I use the ApplyClientBehavior method to add my inspector to the runtime. My implementation looks like the following:
public class ClientMessageInspector : IClientMessageInspector, IEndpointBehavior
{
#region IClientMessageInspector Members
public void AfterReceiveReply(ref Message reply, object correlationState)
{
throw new Exception("The method or operation is not implemented.");
}
public object BeforeSendRequest(ref Message request, IClientChannel channel)
{
object temp = Thread.GetData(Thread.GetNamedDataSlot("CorrelationId"));
if (temp != null)
{
Guid correlationId = (Guid)temp;
MessageHeader<Guid> messageHeader = new MessageHeader<Guid>(correlationId, false, "http://example.com/asyncmsmqservice", true);
MessageHeader untypedHeader = messageHeader.GetUntypedHeader("CorrelationId", "urn:Correlation");
request.Headers.Add(untypedHeader);
}
return null;
}
#endregion
#region IEndpointBehavior Members
public void AddBindingParameters(ServiceEndpoint endpoint, BindingParameterCollection bindingParameters)
{
}
public void ApplyClientBehavior(ServiceEndpoint endpoint, ClientRuntime clientRuntime)
{
clientRuntime.MessageInspectors.Add(this);
}
public void ApplyDispatchBehavior(ServiceEndpoint endpoint, EndpointDispatcher endpointDispatcher)
{
}
public void Validate(ServiceEndpoint endpoint)
{
}
#endregion
}
Once I had the inspector code written, I just needed to add it to the enpoint in my channel factories in both server and client. I have then modified the code to initialise the factories call a method like the following:
static ChannelFactory<IServiceContract> InitializeFactory()
{
var channelFactory = new ChannelFactory<IServiceContract>("*");
channelFactory.Endpoint.Behaviors.Add(new ClientMessageInspector());
return channelFactory;
}
Now I had custom headers getting in and out of the messages. Next step was to be able to tell what value went in the custom header, as I wanted the correlation ID to be set by the application and not automatically by the inspector. For that effect, I have set the correlation ID in the data context of the current thread. As the inspector gets called in the same thread that is making the request it would work, so I changed the code in the inspector to look like the following:
object temp = Thread.GetData(Thread.GetNamedDataSlot("CorrelationId"));
if (temp != null)
{
Guid correlationId = (Guid)temp;
// ...
}
And added the following code to the code that calls the services:
private static Guid InitializeCorrelationId()
{
Guid correlationId = Guid.NewGuid();
Thread.SetData(Thread.GetNamedDataSlot("CorrelationId"), correlationId);
return correlationId;
}
Cool, now I had my own Guids going back and forth to the server. There was only one thing missing: How to hold the thread that initiated the correlation running and then make it continue only when I got the response for that correlation? A custom WaitHandle was the answer. I implemented a custom wait handle that looked like the following:
public class CorrelatedWaitHandle : EventWaitHandle
{
private Guid _correlationId;
private object _contextData;
public CorrelatedWaitHandle(Guid correlationId, bool initialState)
: base(initialState, EventResetMode.AutoReset)
{
_correlationId = correlationId;
}
public Guid CorrelationId
{
get { return _correlationId; }
}
public object ContextData
{
get { return _contextData; }
set { _contextData = value; }
}
}
Then I changed the calling code to keep a collection of these handles (each thread waiting for a response would have its own handle):
//...
var waitHandle = _waitHandles.CreateHandle(correlationId);
waitHandle.WaitOne();
Console.WriteLine("Received '{0}' from the server.", waitHandle.ContextData.ToString());
//...
And finally, using the extension methods functionality in C# 3.0 I could add a couple of convenience methods to the collection itself by writing the following class:
public static class CorrelatedWaitHandleListExtension
{
public static CorrelatedWaitHandle CreateHandle(this List<CorrelatedWaitHandle> list, Guid correlationId)
{
CorrelatedWaitHandle handle = new CorrelatedWaitHandle(correlationId, false);
lock (list)
list.Add(handle);
return handle;
}
public static CorrelatedWaitHandle SetHandle(this List<CorrelatedWaitHandle> list, Guid correlationId)
{
return SetHandle(list, correlationId, null);
}
public static CorrelatedWaitHandle SetHandle(this List<CorrelatedWaitHandle> list, Guid correlationId, object contextData)
{
CorrelatedWaitHandle handle = (from h in list
where h.CorrelationId == correlationId
select h).FirstOrDefault();
if (handle != null)
{
lock (list)
list.Remove(handle);
handle.ContextData = contextData;
handle.Set();
}
return handle;
}
}
In the end my solution worked and my correlations were doing exactly what I wanted. I also moved the return address to a message header so that my service contract interfaces didn’t have anything other than business data. I didn’t want the correlations to be persistent (last a client restart) as I want to ignore any messages that was received before the client restarted if that happened. I’m now in the process of performance testing the whole thing and looking for pitfalls (I’m mainly concerned about keeping many wait handles at once in memory). If you see any problems or has any suggestions, please give me a shout.
No comments:
Post a Comment