A beginner's guide to queuing with WCF and MSMQ showing bi-directional client correspondance

• Download the source projects - 94.6 KB

Objective

This article arose from a learning exercise on my part where the aim was to use WCF to harness MSMSQ for a system where purchase and consumption data needs to be shared in real time between differnt installations. It is a scenario where every client is both a publisher and a subscriber. Queuing a response is where most of us who are less than expert in the fine technical details get lost with the MSMQ/WCF examples avaiable at the time of writing. They range from too complex to follow, use non standard constructs, or simply dont apply queuing properly in the response element.

My aim in this article is to provide a working sample that is easy to follow and replicate - even if not necessarily easy to understand.

This article has three primary parts, the conventional sending of messages from the client to the service, placing data (in this case addressing) on the outbound message header *to help address messages back to the client) and finally sending messages from the service to the client.

Example Origin - Why MSMSQ?

Because of the nature of my system, I cannot guarantee that the clients and service will always be able to reach each other, similarly when a client comes online, it needs to be able to pick up any service oriented broadcasts that made while it was offline.

In a deployment where the target queue is on an unavailable remote machine, MSMQ will automatically - with no effort on your part - hold the messages locally and forward them to the target machine when it becomes available.[ROGERS]

My original example was a key stepping stone in building my knowledgebase but was dependant on both client and service being online simultaneously and so was unsuitable for any actual use. This example represents the model in plan to develop and deploy.

Here, each client creates its own queue for inbound messages at run time, and places the address of that queue on the join request to the service. The service creates a dictionary holding the proxy for each client using the client name as the key. Clients write messages and tick recipients on a list then send the messages onto the service queue. The service reads this queue, and examines the address list on each message. Using the dictionary of stored proxies, it is able to queue those messages for each ontended client.

NOTE that if the service goes down, you will need to re register the users to avoid an error - this is because I have not implemented code to save and reload the dictionary.

However while the service is up, you will be able to knock clients offline, send them messages from other clients, and see those messages come through when that client comes back online.

This is very similar to publish and subscribe, but not the same in that there is no dedicated publisher or subscriber. In my example every client is both a publisher and a subscriber. Tbe service meanwhile, is no more than a relay station.

My previous WCF offering had its origins in two excellent sources of inspiration, both of which were essential in achieving its objective:

• [TROELSEN] - Chapter 25 of "Pro C# 2010 and the .NET 4 Platform (Fifth Edition)" By Andrew Troelsen
• [BARNES] - "WCF: Duplex Operations and UI Threads" by Jeff Barnes, right here on CodeProject

On this occasion there I found no standout example to base my effort on, so what you have on this occasion is largely my own work with a mosaic of minor contributions from an array of sources. I believe I have listed all of them at the end of this article, but apologies in advance if there are any omissions.

The original plan was to adapt the source from my GPH_QuickMessageService showing how to convert it from connected to queued - but this proved a bridge to far on top of learning how to queue messages and responses. Before being even able to embark on this excercise I had to put together a trivial example that where the client placed a hardcoded message on a queue that was read and written to the console by the service. [LOWY] was instrumental in quiding me through that steppping stone. Pointless in its own right, but essential in setting me off towards a meaningful bi directional queued solution.

The queuing method I employ is transactional and I will point it out at various stages in the code.

If this is your first experience of WCF then I suggest reading my initial article, and the seminal pieces I drew from as quoted in it.

NOTE: For the avoidance of doubt - this is a C# example.

Before you begin...

Now is a good time to go to the Control Panel, Programs, Turn Windows Feature On or Off and make sure you have MSMQ active.

Sending messages from the Clients to the Service

First up, I will detail the code I used to queue messages from the client to the service. There is little unusual in it, and there are plenty of other examples about. Once complete we will revisit the entire code set and weave in the logic required to send messages from the service to the clients.

The Service

Open a new solution for a console application, I called mine GPH_QueuedMessageService. Rename the auto-generated Program.cs to something more relevant. I chose GPH_QueuedMessageHost.cs. Park that for the moment and add a new class library project to the solution - this will contain the service contract, and the name should reflect this. I chose GPH_QueuedMessageContract. Rename the auto-generated Namespace and class names to better reflect the work you are doing. We will examine this contract first:

The Contract: GPH_QueuedMessageContract

Start by adding the following references to GPH_QueuedMessageContract.cs:

using System.ServiceModel;
using System.Transactions;  // For Transaction Scope
using System.Messaging;     //Access Messaging
using System.Runtime.Serialization; //DataContract

I have adopted the source file name prefixed by "N" as my namespace name (NGPH_QueuedMessageContract).

Within the namespace, you will find a standard service contract:

[ServiceContract(
Name = "GPH_QueuedService"
)]

public interface IGPH_QueuedService
{
    [OperationContract(IsOneWay = true)]
    void RegisterUser(string arg_Username);
    [OperationContract(IsOneWay = true)]
    void ReceiveMessage(string userName, List<string> addressList, string userMessage);
    [OperationContract(IsOneWay = true)]
    void RemoveUser(string arg_Username);
} 
</string>

For those of you who have examined my earlier connected WCF example, this will be very familiar, all three methods have a similar signature, with two of them given new names to reflect differences in how they will behave in a queued world. Because this is a disconnected example, there cannot be anything returned to the client via the method type - the client could be gone offline when the message is processed - so the return type has to be void, and all are now adorned with the OneWay attribute.

This time there will be no callback interface. After I have described how the service receives queued messages, we will go back through the code and write in the logic required to distribute those messages to the addressed clients.

As with any other WCF implementation, the class in the contract module uses the interface to pick up the contract:

public class CGPH_QueuedMessageContract : IGPH_QueuedService

In common with my previous example, I am holding a list of subscriber names, and a dictionary including both the subscriber names and their proxies (strictly as per my intro they are now more users than subscribers - theres legacy for you).

private static List<string> m_SubscriberList = new List<string>();
private static Dictionary<string, GPH_QueuedMessageContract.ServiceReference1.GPH_InboundMessageHandlerClient> m_NotifyList =
   new Dictionary<string, GPH_QueuedMessageContract.ServiceReference1.GPH_InboundMessageHandlerClient>();// Default Constructor</string></string>

Registering a user.

The most striking feature of the method shown here is the OperationBehaviour attribute. The method could not be used on a transaction based queue without this attribute.

For the moment when a user registers, all I will show is that a 'join request' has been made by putting out a console message:

[OperationBehavior(TransactionScopeRequired = true, TransactionAutoComplete = true)]
public void RegisterUser(string arg_Username)
{
    Console.WriteLine("Got Message");
}

I have taken that decision, because the logic that appears here is used to record addresses that will be used to dispatch messages to the clients, so I will take you through it when we look at sending messages from the serivce to them.

Receive a message.

With the exception of the behaviour attribute added, there is nothing special required in receiving a message. Again I am using a console writeline to show that it has arrived.

[OperationBehavior(TransactionScopeRequired = true, TransactionAutoComplete = true)]
public void ReceiveMessage(string arg_userName, List<string> arg_addressList, string arg_userMessage)
{
    Console.WriteLine("Message [{0}] from [{1}]", arg_userMessage, arg_userName);
}</string>

Deregistering a user.

When a deregistration message comes in, this method is used to remove that user from the subscriber list and dictionary so we will see much more of it when we look at messages to the client. For now it simply signals receipt of a deregistration message with a writeline:

[OperationBehavior(TransactionScopeRequired = true, TransactionAutoComplete = true)]
public void RemoveUser(string arg_Username)
{
    Console.WriteLine("Got Removal Message from [{0}]", arg_Username);
}

The Host: GPH_QueuedMessageHost

If you started by creating GPH_QueuedMessageHost and adding the contract library then skip the next paragraph.

But if the only module in your host project is the contract library then now add a new project to your solution (Ctrl-Shift-N) using the Console Application template. I named mine GPH_QueuedMessageHost. Make sure "Create Directory for Solution" is unticked, and select "Add to Solution" on the dropdown it you want to follow my lead and include it in the same solution as the contract.

Once you get your new module generated, I suggest renaming the class name from the default program to something more relevant. I chose GPH_QueuedMessageHost.

Regardless of how or when you created your host module, it will need these references to perform its tasks:

using System.ServiceModel;
using System.Configuration; //Access app.config
using System.Messaging;     //Access Messaging
using NGPH_QueuedMessageContract;

The Main method commences showing the influence of [LOWY] - if the queue does not exist create it programmatically.

// Get MSMQ queue name from app settings in configuration
string queueName = ConfigurationManager.AppSettings["queueName"];
// Create the transacted MSMQ queue if necessary
if (!MessageQueue.Exists(queueName))
    MessageQueue.Create(queueName, true);//Creates a transactional queue.

Then I bring in the base address from App.Config:

string baseAddress = ConfigurationManager.AppSettings["baseAddress"]; 

This is the address that is used to listen for WS-MetaDataExchange requests and will be employed when I generate the proxy for the service on the client side.

All that is left to complete the service is the code to get it up and running:

using (ServiceHost serviceHost = new ServiceHost(typeof(CGPH_QueuedMessageContract), new Uri(baseAddress)))
{
    // Open the host and start listening for incoming messages.
    serviceHost.Open();
    // Keep the service running until the Enter key is pressed.
    Console.WriteLine("The service is ready.");
    Console.WriteLine("Press the Enter key to terminate service.");
    Console.ReadLine();
}

Note that when using creates the context for the service, both the contract and the base address are used as parameters.

The Configuration: App.Config

Before we attempt to compile up our service it needs a configuration file (App.Config).

Right click on the host project name and choose "Add, New Item" then highlight Application Configuration File as illustrated.

Accept the default name and click OK.

The appSettings will provide both the name of the queue and its address:

<appSettings>
    <!-- use appSetting to configure MSMQ queue name -->
    <add key="queueName" value=".\private$\GPH_QueuedServiceQueue"/>
    <add key="baseAddress" value="http://localhost:8080/GPH_QueuedService"/>
</appSettings>

Within the <system.serviceModel> tags we have services, bindings and behaviours:

<services> pulls in all the components required to make the service work. It is qualified by a behavior configuration, the service name, endpoint address, binding, binding configuration, contract and endpoint contract.

Pay particular attention to how the service name is composed (Namespace.Class) and the composition of the contract attribute (NGPH_QueuedMessageContract.IGPH_QueuedService). Getting these wrong will lead to runtime errors that are difficult to spot.

<services>
      <service
      behaviorConfiguration="GPH_QueuedServiceBehaviors"
      name="NGPH_QueuedMessageContract.CGPH_QueuedMessageContract">
        <!--Namespace.Class-->
        <endpoint address="net.msmq://localhost/private/GPH_QueuedServiceQueue"
        binding="netMsmqBinding"
        bindingConfiguration="DomainlessMsmqBinding"
        contract="NGPH_QueuedMessageContract.IGPH_QueuedService"/>
        <!--Namespace.Interface-->
        <!-- Add the following endpoint. -->
        <!-- Note: your service must have an http base address to add this endpoint. -->
        <endpoint contract="IMetadataExchange" binding="mexHttpBinding" address="mex" />
      </service>
</services>

The binding used is netMsmqBinding. This is what tells the service that it will be looking at a queue. Its binding name, DomainlessMsmqBinding is also the service binding configuration and was chosen following an example in [PATHAK]. All that happens in this part of the the config file is that I instruct the service not to expect any authentication or protection on messages. [LOWY] devotes an entire chapter to security.

<bindings>
      <netMsmqBinding>
        <binding name="DomainlessMsmqBinding">
          <security>
            <transport msmqAuthenticationMode="None" msmqProtectionLevel="None" />
          </security>
        </binding>
      </netMsmqBinding>
</bindings>

There is very little happening in the service behaviour. This is a default that I have used since [TROELSEN], but also features in [LOWY] and [PATHAK] among others.

<behaviors>
      <serviceBehaviors>
        <behavior name="GPH_QueuedServiceBehaviors">
          <!-- Add the following element to your service behavior configuration. -->
          <serviceMetadata httpGetEnabled="true"/>
        </behavior>
      </serviceBehaviors>
</behaviors>

The service is now ready to read messages from its queue. At this point in the development of the code that accompanies the article, I was able to compile and launch the service, allowing me to commence work on the client.

The Client

As in the case of the service, the client will both send and receive messages, but in this section I will concentrate on the conventional getting messages to the service, and deal with the opposite direction later.

Create a new windows forms application in a new solution. I choose GPH_QueuedMessageClient for mine and renamed Form1 to MessageForm. The form should be laid out as per my previous WCF example, and as illustrated by the three instances in the image that heads this article.

Add a form loading event, a form closing event, click events on each the buttons, and a text changed event on the Username textbox. Then add the following references to the code behind your form. In my example its GPH_QueueMessageClientForm.cs:

using System.ServiceModel;
using System.Messaging;
using System.Transactions;

You will also need to add a service reference in order to create a proxy for the service on the client side. Right click on the project name and choose Add Service Reference. You will need the service address from the service App.Config. It is on the line with baseAddress in appSettings: http://localhost:8080/GPH_QueuedService. Insert the service address in the address box as shown.

After you see the message telling you of 1 service found at the address, you can click OK. If you do not get this message, then the service may not be running, there could be an error in your address, or a error in the endpoint definition. For example if the contract name has a typo it could all look perfect, but you will get a message at this step telling you there is no endpoint exposed!

A new using statement will join your list of references:

using GPH_QueuedMessageClient.ServiceReference1;

Declare a variable for the proxy in the message form class:

private static ServiceReference1.GPH_QueuedServiceClient m_Proxy;

Initialization

Initialize the proxy in the class constructor:

public MessageForm()
{
    InitializeComponent();
    m_Proxy = new ServiceReference1.GPH_QueuedServiceClient();
}

Loading the From

The MessageForm_Load has no WCF specific code, just some instructions to improve usability of the application:

// Initialize the fields / buttons
this.btnJoin.Enabled = false;
this.btnSend.Enabled = false;
this.btnLeave.Enabled = false;
this.btnExit.Enabled = true;
this.txtMessageOutbound.Enabled = false;

// Initial eventhandlers
this.txtUsername.TextChanged += new EventHandler(txtUsername_TextChanged);
this.FormClosing += new FormClosingEventHandler(MessageForm_FormClosing);

Join

The btnJoin_Click has a lot of code dedicated to providing data for receiving messages. More on that later. To send a join message to the service, this is all that is required:

m_Proxy.RegisterUser(txtUsername.Text);

There is also some form handling code to improve the behaviour of the application that is of no relevance to WCF:

// change the button states
this.btnJoin.Enabled = false;
this.btnSend.Enabled = true;
this.btnLeave.Enabled = true;
this.txtMessageOutbound.Enabled = true;

Exit

The btnExit_Click event does no more than fire the closing event:

this.Close();

Closing the form.

The MessageForm_FormClosing makes sure the proxy is closed with the form:

m_Proxy.Close();

Entering a Username

The txtUsername_TextChanged event only serves to enable the Join button where a the username box has a value:

if (this.txtUsername.Text != String.Empty)
{
    this.btnJoin.Enabled = true;
}

Sending a Message

In our first look at btnSend_Click we will concentrate on the act of sending a message to the service, ignoring the addressing functionality because this is not usable until we can receive a list of subscribers from the service.

This is the code:

txtMessageOutbound.Enabled = false;
m_Proxy.ReceiveMessage(this.txtUsername.Text, addressList, this.txtMessageOutbound.Text);
txtMessageOutbound.Clear();
txtMessageOutbound.Enabled = true;

The only required line in this sample is the one commencing m_Proxy...., the others serve only to disable the text box, clear it and re-enable it. The reason I have gone for disabling it is because queuing is slow on my machine and I don't want to muddy the waters at this early stage by firing multiple messages at the service in quick succession.

The Leave Event

The btnLeave_Click event is used to signal to the service that this user is to come off the subscriber list and to delete its queue for inbound messages on the assumption that the user is finished using the application and does not wish to receive any messages while off line. NOTE: Messages will continue to be queued for users who do not click Leave but close the applicaiton using Exit. This event also resets the form.

m_Proxy.RemoveUser(this.txtUsername.Text);
this.btnJoin.Enabled = false;
this.btnSend.Enabled = false;
this.btnLeave.Enabled = false;
this.btnExit.Enabled = true;
this.txtMessageOutbound.Enabled = false;
if (MessageQueue.Exists(m_queueName))
    MessageQueue.Delete(m_queueName);

Using the message Header

I am following the advice of [LOWY] and using the message header for the exchange of technical data like the inbound (response) address or error codes. This keeps the message body set aside for the business data exchanged by the application. Apart from lending his wisdom [LOWY] was of limited further use because he implemented it using contexts firmly interwoven with his generic ServiceModelEx offering and I was unable to distill out all of what I needed for a dedicated example.

The Service

What I did learn from [LOWY] was that I could pass a dedicated class on the message header as a Data Contract. This belongs in the module with the Service Contract, so off I went to GPH_QueuedMessageContract to add a new class called PublishToDataContract:

[DataContract(Name = "PublishToDataContract")]
public class PublishToDataContract
{
    [DataMember]
    public string PublishToAddress { get; set; }
    [DataMember]
    public readonly string FaultAddress; // Future use
    [DataMember]
    public readonly string MethodId; // Future use
}

This time its known as a DataContract rather than a ServiceContract. Any member that I intend to see on the client side has to be tagged with the DataMember attribute.

So I compiled it up, consulted [LOWY], [PATHAK] and a few others at length, but I could not access the data contract class members on the client side after updating the service reference.

Exposing a 'Null' Method

The solution, a gem of a workaround was given to me as an answer to a question right here on CP!. I consider it Exposing a 'Null' Method. This is why. Add a new entry in the IGPH_QueuedService interface:

[OperationContract(IsOneWay = true)]
void ExposeContract(PublishToDataContract arg_publish_details);

In the CGPH_QueuedMessageContract class, declare a new variable using the type of the new data contract class:

PublishToDataContract m_PublishToDetails = new PublishToDataContract();

Also in this class, implement the ExposeContract method recently added to the interface:

public void ExposeContract(PublishToDataContract arg_publish_details)
{
    ;
}

As you can see it does absolutely nothing in itself. But if you restart the service, go to the client side and re-import the service reference, the PublishToDataContract will now be available for use - and we will be making no mention of ExposeContract on the client side.

Before heading off to the client side, we will add logic to the pull the address of the header, and display it in the console window.

This is the code to get the header from the message and write it out:

try
{
    m_PublishToDetails = OperationContext.Current.IncomingMessageHeaders.GetHeader<PublishToDataContract>(
        "PublishToDataContract", "NGPH_QueuedMessageContract");
}
catch (Exception e)
{
    Console.WriteLine("Exception [{0}] reading the header", e.Message);
    Console.WriteLine("Header 0: {0}",OperationContext.Current.IncomingMessageHeaders[0].ToString());
    return;
}
try
{
    Console.WriteLine("Join Request from Queue: {0} via hdr passing addr: {1}", 
                      arg_Username, m_PublishToDetails.PublishToAddress);
    m_message = "Join Request from Queue: " 
                     + arg_Username + " via address " + m_PublishToDetails.PublishToAddress;
}
catch (Exception e)
{
    Console.WriteLine("Exception [{0}] displaying the header details", e.Message);
    return;
}

The Client

As you have just seen, you will need to update the service reference to access the Data Contract that we will use to pass the client address to the service. Right click on ServiceReference1 and choose Update Service Reference from the popup. The service will need to be running for this to succeed.

The client will now need to reference configuration:

using System.Configuration; //Access app.config

In the btnJoin_Click event, get the root of the endpoint address from the App.Config file and add the current username to it for uniqueness:

string endpointAddressRoot = ConfigurationManager.AppSettings["endpointAddressRoot"];
string strEndpointAddress = endpointAddressRoot + txtUsername.Text;

I have chosen this approach because it allows me to have multiple dynamically addressed clients on the same machine.

Define an instance of the new data contract class:

PublishToDataContract PublishTo;

I followed [LOWY] to populate this instance and get it onto the message header:

PublishTo = new PublishToDataContract();
PublishTo.PublishToAddress = strEndpointAddress;
MessageHeader<PublishToDataContract> numberHeader = 
         new MessageHeader<PublishToDataContract>(PublishTo);

Again, following [LOWY], the proxy call to RegisterUser now has to go onto an inner channel so that the message header can be incorporated I believe. This is how it is now wrapped:

using (OperationContextScope contextScope = new OperationContextScope(m_Proxy.InnerChannel))
{
    try
    {
        OperationContext.Current.OutgoingMessageHeaders.Add(
            numberHeader.GetUntypedHeader("PublishToDataContract", "NGPH_QueuedMessageContract"));
    }
    catch (Exception Ex)
    {
        MessageBox.Show("Exception: {0}", Ex.Message);
    }
    m_Proxy.RegisterUser(txtUsername.Text);
}

We also need to make a change to the client side app.config, adding an appSettings tag set and creating a key to the root of the endpoint address:

<appSettings>
    <add key="endpointAddressRoot" value="net.msmq://localhost/private/GPH_InboundClientQueue_"/>
</appSettings>

We are now ready to start coding so that we can receive messages on the client side from the service.

Receiving messages from the Service on the Client form

The better examples currently available achieve this by adding service code to the client and client code the service. I will be following this pattern.

The Client as a pseudo-Service

Creating the contract

The first thing the client needs to do to take on 'service' like behaviour is to implement a contract. Add a new class library project to the client solution. I called mine GPH_InboundMessageContract and renamed the autogenerated class1 as CGPH_InboundMessageHandler. I have also replaced the auto-generated namespace name, choosing to call mine NGPH_InboundMessageContract.

The contract module will require two references:

using System.ServiceModel;
using System.Messaging;     //Access Messaging

This is a simple client, it will only ever recieve two types of messages, and this is reflected in the service contract:

[ServiceContract]
public interface IGPH_InboundMessageHandler
{
    [OperationContract(IsOneWay = true)]
    void OnRegistration(List<string> arg_SubscriberList);
    [OperationContract(IsOneWay = true)]
    void OnInboundMessage(string arg_User, string arg_Message);
    //More operations
}</string>

In keeping with any WCF service definition, the class implements the contract interface:

public class CGPH_InboundMessageHandler : IGPH_InboundMessageHandler
{
	.
	.
	.
}

Diversion - firing events

Its all very well receiving messages, and if I simply wanted to make a database change or write to disk based on those messages, then I could just make a call from here. But no, in this example I want to show the inbound data in the form that creates this service. For that I will fire two events, one for each call type. They are basic events and beyond the scope of this article so I will be doing little more than point them out.

Two public static member variables are created to aid in the relay of information back to the form:

public static string m_MessageRecieved;
public static string m_FromUser;

Registration

I create the registration event to accept the list of users that will arrive in from the queue, and a handler for it. Then I use the handler in the method that will fire the event:

public class RegistrationEventArgs : EventArgs
{
    public List<string> evUserList;
    public string CallingMethod;
}
public static event EventHandler<RegistrationEventArgs> RegistrationEvent;

public void SendData(List<string> arg_senduserList)
{
    if (RegistrationEvent != null)
        RegistrationEvent(null, new RegistrationEventArgs());
}

Inbound Messages

The ShowMessage event is created in the same manner as the registration event - I could well have parameterised this to have a single generic event, but that is an exercise for another day...

public class ShowMessageEventArgs : EventArgs
{
    public string evMessage;
    public string CallingMethod;
}
public static event EventHandler<ShowMessageEventArgs> ShowMessageEvent;

public void SendMessage(string arg_User, string arg_Message)
{
    if (ShowMessageEvent != null)
        ShowMessageEvent(null, new ShowMessageEventArgs());
}

And back to the Contract

Receiving a Registration Message

Note that the OnRegistration method has an OperationBehavior attribute defined to implement transactions. Functionally all it does is take in a list of subscribers and fires the SendData event to pass that list to the form:

[OperationBehavior(TransactionScopeRequired = true, TransactionAutoComplete = true)]
public void OnRegistration(List<string> arg_SubscriberList)
{
    //MessageBox.Show("Result = " + arg_InboundText, "InboundMessageHandler");
    m_SubscriberList = arg_SubscriberList;
    SendData(arg_SubscriberList);
}

Receiving a regular Message

Similar to OnRegistration, this method takes in a string and a list of users placing their values on the public variables that the form will pick up then fires SendMessage to pass that string to the form:

public void OnInboundMessage(string arg_User, string arg_Message)
{
    m_MessageRecieved = arg_Message;
    m_FromUser = arg_User;
    SendMessage(arg_User,arg_Message);
}

Changes to the client form

The client form has to be changed to create the service that will use this contract. All these changes will take place in the code that handles the client form events - GPH_QueueMessageClientForm in my example.

My first change is to reference the new contract, and ServiceModel.Description for endpoint manipulation:

using System.ServiceModel.Description; // For ServiceEndpoint
using NGPH_InboundMessageContract;

The 'Join' Click Event

In the 'Join' click event (btnJoin_Click) create an inbound (response) queue incorporating the name of the joining user in a root name read from the App.config where that queue does not already exist:

m_queueName = ConfigurationManager.AppSettings["m_queueName"] + txtUsername.Text;
if (!MessageQueue.Exists(m_queueName))
    MessageQueue.Create(m_queueName, true);//Creates a transactional queue.

Next I am creating the binding. Because I want to use dynamic addressing involving the username, I am creating everything at run time. If this is too restrictive, you are free to read in settings from App.Config.

NetMsmqBinding Binding;
Binding = new NetMsmqBinding();
Binding.Security.Transport.MsmqAuthenticationMode = MsmqAuthenticationMode.None;
Binding.Security.Transport.MsmqProtectionLevel = System.Net.Security.ProtectionLevel.None;

Create the endpoint using a combination of an an App.Config entry and the username from the form:

string endpointAddressRoot = ConfigurationManager.AppSettings["endpointAddressRoot"];
string strEndpointAddress = endpointAddressRoot + txtUsername.Text;
EndpointAddress address = new EndpointAddress(strEndpointAddress);

Then I set the base address again involving both the joining username and a root from App.Config, and use it when defining a variable for the client side host:

string baseAddress = ConfigurationManager.AppSettings["baseAddress"];
baseAddress += txtUsername.Text;
ServiceHost host = new ServiceHost(typeof(CGPH_InboundMessageHandler), new Uri(baseAddress));

Add event handlers for the two events fired in the contract to the inbound message handler:

CGPH_InboundMessageHandler.RegistrationEvent
   += new EventHandler<CGPH_InboundMessageHandler.RegistrationEventArgs>(
      CGPH_InboundMessageHandler_RegistrationEvent);
CGPH_InboundMessageHandler.ShowMessageEvent
   += new EventHandler<CGPH_InboundMessageHandler.ShowMessageEventArgs>(
      CGPH_InboundMessageHandler_ShowMessageEvent);

Open a host on the client side, using the inbound message handler, binding and endpoints defined here.

host.AddServiceEndpoint(typeof(IGPH_InboundMessageHandler), Binding, strEndpointAddress);
host.Open();

Handling a Registration Event

When the contract fires a registration event, CGPH_InboundMessageHandler_RegistrationEvent will deal with it on the client side. It does no more than refresh the subscriber check list:

clstSubscriber.Items.Clear();
clstSubscriber.Items.Clear();

foreach (string subscriber in NGPH_InboundMessageContract.CGPH_InboundMessageHandler.m_SubscriberList)
    clstSubscriber.Items.Add(subscriber);

Inbound Messages

Inbound messages events are caught by CGPH_InboundMessageHandler_ShowMessageEvent. It uses the contract variables m_FromUser and m_MessageRecieved, reformatting them and adding them to the inbox text field on the form.

Leaving

When btnLeave_Click is clicked the user response queue will be deleted:

if (MessageQueue.Exists(m_queueName))
    MessageQueue.Delete(m_queueName);

Client side App.Config changes

We gained an App.Config automatically when the client had its service reference created. Now it is time to embellish it further to handle messages coming in to the client.

First we need the roots for our dynamic addressing. These will be key/value pairs in AppSettings:

<appSettings>
    <!-- use appSetting to configure MSMQ queue name -->
    <add key="m_queueName" value=".\private$\GPH_InboundClientQueue_"/>
    <add key="endpointAddressRoot" value="net.msmq://localhost/private/GPH_InboundClientQueue_"/>
    <add key="baseAddress" value="http://localhost:8080/GPH_QueuedInbound"/>
</appSettings>

Theres also a new service entry:

<service
          behaviorConfiguration="GPH_QueuedClientBehaviors"
          name = "NGPH_InboundMessageContract.CGPH_InboundMessageHandler">
</service>

and a new behaviour:

<behavior name="GPH_QueuedClientBehaviors">
    <!-- Add the following element to your service behavior configuration. -->
    <serviceMetadata httpGetEnabled="true"/>
</behavior>

Now compile up the client, run it with admin privileges, key in a username and click 'Join'. I used 'J1' (for Joiner number one) as my initial user, note yours - we will come to rely on it briefly as we get the service up to speed in sending messages back to the clients.

The Service as a pseudo-Client

Here we will look at how the service takes on some client side behaviours creating a proxy for each client side service and using it to dispatch messages to the client.

Program Code changes

The significant changes made are in GPH_QueuedMessageContract to CGPH_QueuedMessageContract in how it implements the contract. But first it needs a service reference. Add it in the same way as you added the service reference to the client, this time use the address of the client you set running. The address of the reference to be added will be http://localhost:8080/GPH_QueuedInboundj1. This flies in the face of dynamic addressing, giving the impression that only known users at design time can hook in to get responses from the service. However it is overcome by overwriting any client specific addresses programmatically when we add response capability to CGPH_QueuedMessageContract.

The CGPH_QueuedMessageContract needs some new member variables.

string m_message;
EndpointAddress m_address;
NetMsmqBinding m_Binding;
private static List<string> m_SubscriberList = new List<string>();


// current plan is for a dictionary of users and proxies where a proxy is created for each user
// on registration and then used for any communication to the user.
// The alternative is to hold the endpoint in the dictionary and create the proxy before each message
// and destroy it afterwards
private static Dictionary<string, 
  GPH_QueuedMessageContract.ServiceReference1.GPH_InboundMessageHandlerClient> m_NotifyList =
  new Dictionary<string, GPH_QueuedMessageContract.ServiceReference1.GPH_InboundMessageHandlerClient>();
  // Default Constructor

As per my Duplex WCF example, I have a standard subscriber list and a dictionary of subscribers referencing on this occasion proxies.

Registering a user.

Make these changes to the RegisterUser method to build the proxy needed to contact the user being registered.

Create a binding:

m_Binding = new NetMsmqBinding();
m_Binding.Security.Transport.MsmqAuthenticationMode = MsmqAuthenticationMode.None;
m_Binding.Security.Transport.MsmqProtectionLevel = System.Net.Security.ProtectionLevel.None;

Pull in the address from the message header - this will override the address hardcoded into the <Client> tag in App.Config

m_address = new EndpointAddress(m_PublishToDetails.PublishToAddress);

Use the new binding and address to create a new proxy for the client being registered:

GPH_QueuedMessageContract.ServiceReference1.GPH_InboundMessageHandlerClient proxy;
//Use an overload to supply the new response address. Store in a key / value table per client
proxy = new GPH_QueuedMessageContract.ServiceReference1.GPH_InboundMessageHandlerClient(m_Binding, m_address);

Store the proxy on the dictionary indexed by the username being registered, and add that username to the subscriber list:

// Store the username and proxy in a lookup table
m_NotifyList.Add(arg_Username, proxy);
m_SubscriberList.Add(arg_Username);

Fire the message distributor method to inform all clients that this user is now registered, supplying an up to date subscriber list:

MessageDistributor(arg_Username, m_SubscriberList, " has joined the converstation",1);

Deregistering and removing a user.

When a user clicks the 'Leave' button on the client side, this triggers the RemoveUser on the service.

First the proxy is retrieved from the dictionary:

GPH_QueuedMessageContract.ServiceReference1.GPH_InboundMessageHandlerClient proxy = m_NotifyList[arg_Username];

The username is removed from the subscriber list and its dictionary entry is also removed.

// Remove the username and proxy from the lookup table and subscriber list.
m_NotifyList.Remove(arg_Username);
m_SubscriberList.Remove(arg_Username);

Then I close the user's proxy:

proxy.Close();

Finally I use MessageDistributor to inform the remaining users that this one has left the conversation, accompanied by an updated subscriber list.

MessageDistributor(arg_Username, m_SubscriberList, " has left the conversation",1);

Recieving a message.

Conisdering that this effort is all about the distribution of messages between clients this method has very litte going on, choosing instead to hand the task over to the MessageDistributor method:

MessageDistributor(arg_userName, arg_addressList, arg_userMessage, 0);

Message Distribution

The MessageDistributor method has been handed the responsibly of addressing messages to the different clients. It takes the username triggering the message, the current address list, the message itself and an indicator as parameters. This indicator will be 0 to process a message from a user only, and 1 when a new distribution list with preset message from service is required.

The logic is wrapped in a foreach loop that will process the address list parameter (the entire list for indicator 0 and as supplied on the message for indicator 1):

foreach (string tmpAddr in arg_addressList)

Every loop iteration is processed within a TransactionScope:

using (TransactionScope scope = new TransactionScope())

Get the proxy from the dictionary:

GPH_QueuedMessageContract.ServiceReference1.GPH_InboundMessageHandlerClient proxy = m_NotifyList[tmpAddr];

Write message to queue, noting that when the indicator is 1, there are two messages to be queued. I could take advantage of transaction scope to roll this back if the second one failed - that is an exercise for another day:

if (arg_ind == 1)
{
    string[] tmpSubscriberList = (string[])m_SubscriberList.ToArray();
    proxy.OnRegistration(tmpSubscriberList); //Send back a list of current subscribers
}
Console.WriteLine("Dispatching message to: [{0}]", proxy.Endpoint.Address);
proxy.OnInboundMessage(arg_userName, arg_userMessage);

The last task is to close the scope:

scope.Complete();

Config Changes

The process of adding a service reference will create an app.config on the contract library - where the application can not see it. The <Client> and <netMsmqBinding> should be copied to App.config on the host project, (GPH_QueuedMessageHost in this example). The hardcoded endpoint address is a byproduct of using the service reference wizard, but is not used in the code.

Looking at your Queues.

You can see the queues on your machine through clicking the start button and right clicking on computer:

Click Manage on the resultant popup shown above. This will open the computer management dialog where you will need to expand Services and Applications then Message Queuing followed by Private Queues and widen the left side pane until you have something like this:

Here is an example of a message queued but not processed:

Double build

When you are using multiple projects in a single solution, you may get link errors because one cannot find the other which dissappear if you compile twice in quick succession - When this happens use the Project Dependecies in the Project menu to correct it.

Bibliography

• [TROELSEN] - Chapter 25 of "Pro C# 2010 and the .NET 4 Platform (Fifth Edition)" By Andrew Troelsen
• [LOWY] - "Programming WCF Services" by Juval Lowy
• [PATHAK] - "Pro WCF 4 Practical Microsoft SOA Implementation" by Nishith Pathak
• [NARAYAN] - "Sending and receiving message in MSMQ using C#" by SheoNarayan
• [HOLLANDER] - "Building a Pub/Sub Message Bus with WCF and MSMQ" by Tom Hollander
• [VANDIEST] - "Create a WCF service and client, using Msmq, programmatically" by Geoffrey Vandiest
• [KUMAR] - "Data Contract" by Saravankumar
• [MSDN] - "How to: Create a Service Endpoint in Code" by MSDN
• [VDSTELT] - "WCF and MSMQ" by Dennis van der Stelt
• [HALABAI] - "WCF Queued Messaging" by Mohamad Halabai
• [DORIER] - "WCF: Duplex MSMQ" by Nicholas Dorier
• [PAUL] - "Topic-based publish/subscribe design pattern implementation in c#-Part II (Using WCF)." by Razan Paul
• [ROGERS] - "MSMQ and WCF (client and service on different machines)" by Will Rogers

History

2013-01-01 - V1.0 - Initial submission.