NServiceBus (NSB) 101: Publish and Subscribe

Here  are my first interactions with the NSerivce Bus (NSB)

==================================================

NSB – Setup

==================================================

1. Downloaded the code (www.nservicebus.com)
   
2. Without looking into the code – I clicked the RunMeFirst.bat file
   This configured DTC, MSMQ and relevant performance counters on my machine (required a system restart)
   Here is how my console looked like:
   I ran it through the command prompt to visualize any errors if reported
3. Open the pub-sub sample located under
   Location: …\.net 4.0\samples\PubSub\PubSub.sln
   Solution Structure:
   

 The solution sample consists of four projects. Lets look into these projects and find out the workings of a publish and subcribe model in the NSB.

==================================================

Project: MyMessages

==================================================

This is the project where I define all my contracts. The important thing to note is that all messages in NServiceBus must ultimately inherit from the IMessage interface. This class defines two message types

1. IEvent: Interface that subscribes to IMessage
2. EventMessage: Classes that implements the above interface  (IEvent)

IMessage is an NSB type interface. It is a marker interface to indicate that a class is a message suitable for transmission and handling by an NServiceBus

Note: I added the SequenceNo field to IEvent class sample to track down messages by an integer field- for simplicity purposes

The above fields would be populated by the publisher and fed into the NSB, where the subscriber would receive the message and handle it.

==================================================

Project: MyPublisher

==================================================

The project has the following files in it:

The MyPublisher project (a console application) runs as a process that will be publishing messages. Project properties screen shot is as follows:

 

This project consists of three classes (described as follows). I’ll start by covering the theory bits ..

1. EndPointConfig: Inherits from  two interfaces (boiler plate code)
   a.     IConfigureThisEndpoint: Indicates that the implementing class will specify configuration.
   b.     AsA_Publisher: Indicates this endpoint is a publisher.  The provided sample file does not implement both interfaces (blank)
    
2. ServerEndpoint: Implements IWantToRunAtStartup and contains an interface for IBus
   a.     IWantToRunAtStartup: Implementers will be invoked when the endpoint starts up. Dependency injection is provided for these types. 
            It implements two methods for the interface
           i.     Run(): Method called at startup. This is where the MyMessage type is populated.
           ii.     Stop(): Method called on shutdown.                
   IBus: Defines a bus to be used with NServiceBus. (Whatever that means!!) .
3. SubscriptionAuthorizer: Inherits from a single interface (boiler plate code)
   a.      IAuthorizeSubscriptions: Implementer will be called by the infrastructure in order to authorize subscribe and unsubscribe requests from other endpoints. Infrastructure automatically registers one implementing type in the container as a singleton.               
   It implements two methods for the interface
                  i.     AuthorizeSubscribe():Return true if the client endpoint is to be allowed to subscribe to the given message type
                  ii.     AuthorizeUnsubscribe():Return true if the client endpoint is to be allowed to unsubscribe to the given message type The provided sample class returns true for both methods 
   … and two configuration files
4. App.Config: Contains two sections
   a.      MsmqTransportConfig: Used to configure the MSMQ parameters. Has the following key properties:
   → InputQueue: The name of the queue on the local machine from which this process will be receiving messages
    → ErrorQueue: The name of the queue to which messages will be transferred if they could not be processed successfully b.      UnicastBusConfig: (This config section can be ignored) Used to configure the bus (more control on configuration). This is the place where you need to tell the Bus what messages you will be allowing for the current process and what endpoint queue you will be using. Providing you have MSMQ installed on your machine NServiceBuscreates this queue for you. → MessageEndpointMappings: Contains the mappings from message types (or groups of them) to endpoints
   and has the following key properties in MessageEndpointMappings section:
             → Messages: A string defining the message assembly, or single message type
             → Endpoint: The queue on which messages are destined towards (format: queue@machine) 
   This tells NServiceBus that all messages in the MyMessages assembly should be routed to the queue called MyPublisherInputQueueon the local machine.Comment: I “think”, the unicastBusConfig would be required if multiple data contracts are being used by the publisher. This would explicitly direct message types from a specific type (a contract) to the destined queue. If not, this section can be ignored.
5. NServiceBus.Host.exe.config  (This config section can be ignored)

Contains one keyvalue pair in the appsettings section.

&amp;lt;appSettings&amp;gt;<br />
&amp;lt;add key=&amp;quot;EndpointConfigurationType&amp;quot; value=&amp;quot;MyPublisher.EndpointConfig, MyPublisher&amp;quot;/<br />
&amp;lt;/appSettings&amp;gt;

where: → EndpointConfigurationType = Used to determine the assembly type of the publisher. Should only be used if you don't want NSB to scan your assemblies OR if you have multiple configurations and want NSB to load a specific one. You can probably remove "EndpointConfigurationType" from  your config.
Value:
MyPublisher.EndpointConfig = Namespace.TypeName,
MyPublisher   = AssemblyName

 ***************************************************************

Project: MyPublisher

¦→ Publishing a msg to the bus

***************************************************************

Let’s look at publishing a message into the bus. I don’t have any queue created on my local machine and would expect the NSB to create two queues

1. MyPublisherInputQueue
2. error

 Once I start a new instance of the project (MyPublisher→Debug → Start new instance), I see three queues created by the bus

Results:

I wasn’t expecting the “mypublisher.endpointconfig_v1.0.0.0_subscription” queue. This queue was created by NSB to manage publishers and subscribers.

An important thing to note here is that the publisher won’t push any message to the queue unless there are relevant subscriptions registered (i.e relevant entries in the mypublisher.endpointconfig_v1.0.0.0_subscription queue). As soon as the relevant subscriber comes on board, the previously sent message by the publisher is matched – an entry created in the endpoint queue and THEN the message is delivered to the subscriber queue.

Lets understand the subscribers to proceed further

==================================================

Project: Subscriber2

==================================================

This project includes a reference to MyMessages project.

The project consists of three classes

1. EndPointConfig: Inherits from  two interfaces (boiler plate code)
   a.      IConfigureThisEndpoint: Indicates that the implementing class will specify configuration.
   b.      AsA_Server: Indicates this endpoint is a server.
   c.      IWantCustomInitialization: If you want to specify your own container or serializer, implement this interface on the class which implements IConfigureThisEndpoint. Implementors will be invoked before the endpoint starts upIt implements one method for initializing the serializer and the UnicastBus
           i.      Init(): Perform initialization logic.

   public void Init()<br />
   {  <br />
   NServiceBus.Configure.With()<br />
   // just to show we can mix and match containers      <br />
   .CastleWindsorBuilder()       <br />
   .XmlSerializer()<br />
     .UnicastBus()<br />
   // just to show we can mix and match containers<br />
   .DoNotAutoSubscribe();   <br />
   }

   For more info on Castle Windsor: Click here, and here

2. Subscriber2Endpoint
   It implements two methods to manage subscriptions manually
                       i.  Run(): Method called at startup.
   Code:

   public void Run()   {<br />
               Bus.Subscribe&amp;lt;IEvent&amp;gt;();        }

   We’re subcribing to all IEvent messages from the MyMessages project.

                     ii.  Stop(): Method called on shutdown. (Contains

   Bus.Unsubscribe<IEvent>();

3. EventMessageHandler
    This class handles for the subscribed IEvent event (step 2 above).
   Code:

   public void Handle(IEvent message) {<br />
          Console.WriteLine(&amp;quot;Subscriber 2 received IEvent with SeqNo &amp;quot;+ message.SequenceNo); }<br />
   

   I’m printing out the Sequence Number property that was previously populated by the Publisher.

4. App.Config
   Similarly to the publisher, the subscriber contains configuration for its receive location. It contains the input queue from which it would receive its messsages (Subscriber2InputQueue).♦♦♦♦♦♦♦♦ Key point: In addition to the MSMQ Transport Config above, the subscriber should send a message to the publisher indicating its desire to subscribe. The only way the subscriber can do this, if they know the publisher’s endpoint.
   
   
   Therefore, we add the UnicastBusConfig entry to tell the publisher, that the subscriber is interested in messages from MyMessages, that arrive on the publisher’s endpoint “MyPublisherInputQueue“

 

==================================================

The Workings: Publish and Subscribe

==================================================

 As I started the MyPublisher project, I noticed the creation of three queues as expected (and previously described above). Similarly, as I started my subscriber, I received an error indicating that the Subscriber2InputQueue hasn’t been created. All these queues were automatically created by the NSB. 

 

Publisher: As I bring up the publisher (Debug → New Instance) and hit the enter key, a message gets sent to the NSB (nothing gets written to any queue – yet). Because there are no awaiting subscribers – the message does not get written to our queue. A message would only reach the queue from a publisher, if there are subscribers registered against it.

Subscriber: As I bring up the subscriber (Debug → New Instance), a subscription (queue message) is created on an intermediary queue called “mypublisher.endpointconfig_v1.0.0.0_subscription”. This was created by the NSB. Once I examine the contents of the queue it appears as follows:

This reads out as, “All endpoint subscriptions for mypublisher  (1) are listed as follows  – under (2). All messages of type (3) that are published by (1) would be deliverd to (2)“

In plain terms, it simply means all messages of type MyMessage.IEvent, would be delivered to the Subscriber2InputQueue queue. An interesting point to note is that  – if we create this subscription manually, there would be no need to configure the UnicastBusConfig for the subscriber !! – food for thought . Adding onto it, what if you could maintain all these configurations inside a database (also possible)

Finally, as I hit the enter key – messages from the publisher are being delivered to the subscriber as expected.

Points of consideration:
The Server process publishes messages by looking through messages in the ‘subscriptions’ queue. The server attempts to match the message type it is currently sending with a subscriber that’s interested in the message.  If the server finds a subscriber, it publishes the message to the subscriber’s queue – otherwise, it ignores the message and continues to the next message. (Ref: Click here)

==================================================

Conclusion

==================================================

Conclusion: My first take on the NSB have found it to be light-weight communicator between .NET endpoints. By endpoints I  mean two .NET interfaces – be it applications, webservices, etc. It is not a replacement for a powerhouse like BizTalk in anyway – but could gel well together with it. NSB can appear as the communication layer between applications, where BizTalk would act as a subscriber communicating with LOB systems and complementing the NSB where it could not meet the requirements.

The NSB falls short on so many grounds, with things like:

→ Rules or content based (XPATH/Context,etc) based communication/routing between publishers and subscribers

→  Configuration of error queues and rules around them within the arrangement of our architecture (Standard way to handle exceptions and rules around the types of exceptions)

→ Replay messages (hitting everything in one exception queue might not be beneficial enough??)

→  Message transformations between subscribers and publishers. What if the message formats between a subscriber and publisher are not compatible?

→ Tracking performance and monitoring services and gathering statistics to track bottlenecks.

→… and many more..

The discussion on NSB vs BizTalk is a long story. The discussion is actually more around capabilities versus distributed communication. I would probably look to write a seperate post to discuss that. For now, all this post aimed at - was a crash course on the pub-sub pattern using the NSB Happy coding ..

 

Advertisement