Short summary of tests performed:

• we tested sending messages, receiving them and request-reply case (receiving a message and replying to it)
• each time the test was performed on 100 000 messages and we measured the time taken
• Nginn-messagebus handles durable and transactional messages only so take it into consideration when comparing with other tools
• We tested local queue performance only (without sending messages to a remote database).

You can find the test details here.

Test cases 1 and 2 – sending messages ¶

sending messages – 1 thread, no batching

sending messages – 1 thread, batches of 10 per transaction

Multithreaded sending messages, 4 threads, batch size 10

Multithreaded sending messages, 4 threads, no batching

Multithreaded sending messages, 8 threads, no batching

Multithreaded sending messages, 8 threads, batch size 10

Test case 3 – receiving messages ¶

receiving messages – 1 thread

receiving messages – 4 threads

receiving messages – 8 threads

receiving messages – 12 threads

Test case 4 – Request/reply ¶

Request-reply, 12 threads

Request-reply, 8 threads

Request-reply, 4 threads

And some final thoughts ¶

• The numbers are very good, both for sending and for receiving. NGinn-messagebus offers very good transactional processing performance,probably better than MSMQ (for transactional and durable messages only)
• Although the tests showed quite good peak performance, they did not tell us anything about sustainable performance – that is, how many messages per second we are able to process when the load is sustained for a long period of time.
• Send performance was very good, but we did not test how fast the data is removed from the queue. The cleanup did not kick in in such short testing.
• We tested local database only, without transferring the messages to a remote database. Such functionality is present in NGinn-messagebus but currently is not very fast because messages are sent one by one. Further optimization is needed here and also a more elaborate testing environment for evaluating the performance.

• It would be very interesting to perform similar tests for SQL Service Broker and for MSMQ and compare the results. I suppose SSB would be not that fast when sending messages but it would be much better in transferring messages to a remote database and might be faster in receiving (because with SSB we don’t have to use polling to retrieve the messages). MSMQ is fast but usually you have to use a distributed transaction when you want to process the message in a transactional way and this kills performance.

Test setup ¶

All tests were performed on a Dell Latitude E6510 laptop with 8 Gb of RAM, Intel Core i7 740QM processor (8 cores reported) and a single 500 GB 5400 RPM hard drive. Operating system was Windows 7 Professional x64. SQL Server 2008 developer edition.
This means that these results will not tell you too much about what performance to expect from nginn-messagebus on a real server machine, but you can expect that the numbers will be much better for real hardware. As soon as I have a decent testing machine I will post updated test results for comparison.

Test 1. Sending messages to local queue ¶

In this scenario we are publishing or sending messages to a local queue. It means that messages are inserted directly into the underlying queue table so what we are really testing is SQL server insert performance. Here’s the test code:

        static void TestSending2()
        {
            var mc = Configure("sql://nginn/MQ_PT1", true);
            IMessageBus mb = mc.Resolve<IMessageBus>();
            Thread.Sleep(1000);
            Console.WriteLine("Starting TestSending2");
            DateTime dt = DateTime.Now;
            int N = 100000;
            int B = 10;
            var to = new TransactionOptions { IsolationLevel = IsolationLevel.ReadCommitted, Timeout = TimeSpan.FromSeconds(30) };
            for (int i = 0; i < N / B; i++)
            {
                using (var sc = new TransactionScope(TransactionScopeOption.Required, to))
                {
                    for (int j = 0; j < B; j++)
                    {
                        mb.Send("sql://nginn/MQ_PT1", new TestMessage1 { Id = (i * B + j).ToString() });
                    }
                    sc.Complete();
                }
            }
            TimeSpan ts = DateTime.Now - dt;
            Console.WriteLine("Sent {0} messages in {1}, {2} msgs/second", N, ts, N / ts.TotalSeconds);
        }

This code sends 100000 messages in a single thread, as individual messages or in batches (B variable is the batch size) to local sql://nginn/MQ_PT1 queue). Each batch is sent as a single transaction. Messages are left in the queue (no one is consuming them during the test). Here are the test results for single message sending (B = 1) and for batches of 10 messages (B = 10). The numbers mean messages sent per second.

Sending single messages:

Sending in batches of 10 messages:

As you can see, batching the messages improves the performance but the improvement is only about 30% compared to sending of single messages.

Test 2: multi-threaded sending of messages ¶

This is basically the same test as the previous one (sending messages to a local queue) but in this case we are sending 100000 messages using multiple threads. The test was performed for 4 and 8 threads and for single messages in transaction vs batches of 10 messages.

Here’s the testing code:

static long _cnt = 0;
static IMessageBus Bus { get; set; }
static int N;
static int B;

/// <summary>
/// batched sending
/// </summary>
static void TestMTSending2(int batchSize, int threads)
{
    var mc = Configure("sql://nginn/MQ_PT1", true);
    Bus = mc.Resolve<IMessageBus>();
    Thread.Sleep(1000);
    Console.WriteLine("Starting TestMTSending2");
    DateTime dt = DateTime.Now;
    N = 100000;
    B = batchSize;
    int T = threads;
    List<Thread> proc = new List<Thread>();
    for (int t = 0; t<T; t++)
    {
        var th = new Thread(new ThreadStart(ThreadedSend));
        proc.Add(th);
    }
    foreach (var th in proc)
    {
        th.Start();
    }
    foreach (var th in proc)
    {
        th.Join();
    }

    TimeSpan ts = DateTime.Now - dt;
    Console.WriteLine("TestMTSending2 sent {0} messages in {1}, {2} msgs/second", N, ts, N / ts.TotalSeconds);
}

static void ThreadedSend()
{
    var to = new TransactionOptions { IsolationLevel = IsolationLevel.ReadCommitted, Timeout = TimeSpan.FromSeconds(30) };
    while (true)
    {
        using (var sc = new TransactionScope(TransactionScopeOption.Required, to))
        {
            for (int j = 0; j < B; j++)
            {
                var i = Interlocked.Increment(ref _cnt);
                if (i >= N) return;
                Bus.Send("sql://nginn/MQ_PT1", new TestMessage1 { Id = i.ToString() });
            }
            sc.Complete();
        }
    }
}

And the results are

4 sending threads, individual messages (B=1):

As you can see, introducing 4 threads boosted the performance 4-fold.

4 sending threads, batches of 10 messages (B=10):

8 sending threads, batches of 10 messages:

8 sending threads, single messages (B=1):

In this case we have ‘peaked’ at about 6500 messages sent/second when sending single messages in 8 threads.

Test 3: Receiving messages ¶

In this test we will be receiving 100000 messages that were sent in previous tests. The received message is delivered to its handler that does nothing (only logs the fact that the message has been delivered). We will be varying the number of receiving threads.

The code:

//this function just waits until all messages are received
static void TestReceiving1()
{
    var mc = Configure("sql://nginn/MQ_PT1", false);
    IMessageBus mb = mc.Resolve<IMessageBus>();
    Console.WriteLine("Now receiving. Enter to exit");
    Console.ReadLine();
}

public class TestMessageHandler1 : IMessageConsumer<TestMessage1>
{
    private static Logger log = LogManager.GetCurrentClassLogger();

    public void Handle(TestMessage1 message)
    {
        log.Debug("Handling {0}", message.Id);
        //do nothing
    }
}

In this case the processing time is recorded in the database (there’s last_processed column in the queue table that contains the time when message was processed). Based on that column we can calculate the overall message receiving performance.

Single threaded receive

4 receiving threads

8 receiving threads

12 receiving threads

Test 4 – Request-Reply scenario ¶

In this test for each received message we will be sending a single reply message. This is basically the same test as (3) but with the addition of a reply.

The code:

public class TestMessage1Reply
{
    public string RequestId { get; set; }
}

public class TestMessageHandler1 : IMessageConsumer<TestMessage1>
{
    private static Logger log = LogManager.GetCurrentClassLogger();

    public IMessageBus MessageBus { get; set; }

    public void Handle(TestMessage1 message)
    {
        log.Debug("Handling {0}", message.Id);
        MessageBus.Send("sql://nginn/MQ_PT2", new TestMessage1Reply { RequestId = message.Id });
    }
}

The only difference is that now we are sending a TestMessage1Reply message when receiving TestMessage1.

8 receiving threads

12 receiving threads

4 receiving threads

1 receiving thread

Conclusions, further optimization ¶

These performance numbers very good considering the hardware used for testing. They show that nginn-messagebus performance should be more than enough for most OLTP applications and that it could be significantly improved with proper hardware setup.
Monitoring the machine resource utilization during tests showed that the main limiting factor was the hard disk write performance. SQL server could perform better if it could write to disk faster – in all tests the disk average disk write queue was between 1 and 6. CPU utilization was at about 50% in most intensive testing.
Summing up, the performance of nginn-messagebus depends almost exclusively on performance of the underlying SQL server database and can be improved with many SQL server optimization techniques.

Exclusions ¶

All these tests were performed with local queues (stored in a local, directly accessible database). If we were sending messages to a remote database (with store & forward in a local db) the performance would be much worse because currently NGinn-messagebus is not very effective in transporting messages between databases. This limitation stems from the fact that nginn-messagebus does not use batching when moving messages to a remote destination and sends each message in a separate transaction. This is probably the most important area to optimize in near future.

What’s it for ¶

NGInn messagebus can be configured to send messages between endpoints using http. It’s not the most efficient message transport method (performance similar to web service calls) but it allows you to send messags across the network when the destination database is not directly accessible. This function is called ‘http message gateway’ and in order to use it you have to enable the http listener in NGinn Messagebus configuration.

Configuration ¶

The following configuration code will enable embedded http server

var ct = MessageBusConfigurator.Begin()
    .SetConnectionStrings(dbConnectionStrings)
    .SetEndpoint(endpointName)
    (...)
    .ConfigureHttpReceiver("http://MyMachine:8090/")
    (...)
    .FinishConfiguration();

The ConfigureHttpReceiver call tells NGinn-mesagebus to listen on port 8090 on local IP associated with MyMachine name. If you need to listen on all IP addresses of the machine you should specify http://+:8090/ as the listener address.

Using the http gateway ¶

Sending messages to remote endpoints with http is as easy as specifying http address as the endpoint name – like that:

IMessageBus bus (...);
bus.Send("http://YourMachine:9080/", new TestMessage { Id = "AB9932" });

If you are sending messages with http you should configure the http listener at your endpoint because any replies to your messages will be sent back also with http.
NGinn-messagebus implements store & forward for all outgoing messages so even if the remote http endpoint is not accessible the Send will succeed and the message will sit in the local database until the receiver becomes available.

Monitoring and remote control ¶

Since we already have the http server it can be used for many other tasks. One of such tasks is monitoring the health of the service.
If you navigate to http://localhost:8090/health/ you will see the information about current status of message bus (and othere services, if present) and you can use this address with automatic monitoring tools like Nagios.
And if you open http://localhost:8090/businfo you will see a simple page with information about current configuration and with an option to pause message processing. There is also a performance statistics view that allows you to collect performance data, but it will be described in another post.

Json RPC ¶

NGinn-messagebus provides a synchronous Json-based RPC mechanism that allows you to call services remotely. As everything in this project it’s based on concept of messages and their handlers.
First – the message. It defines the service contract:

    public class TestMessage1
    {
        public int Id { get; set; }
    }

Second – service message handler. You have to implement IMessageHandlerService interface:

    public class TestService : IMessageHandlerService<TestMessage1>
    {
        public object Handle(TestMessage1 message)
        {
            if (message.Id == 100) throw new Exception("hundred");
            return new TestMessage1 { Id = message.Id + 1 };
        }
    }

And that’s it, we have defined a service that handles TestMessage1 – it’s very similar to defining message handlers, the only difference is that a service can return a value. We don’t define a contract for the return value – the caller is responsible for specifying the expected return type, as we’ll see further.
Before the service can be called you have to register it in the NGinn messagebus container – for example like that:

var cfg = MessageBusConfigurator.Begin()
    .RegisterHttpMessageServicesFromAssembly(typeof(Program).Assembly)
(...)

Calling the services.
Service can be called by sending HTTP POST with a Json message in the body to service handler URL:
http://localhost:9080/call/TestMessage1. Or we can use a helper class from the NGinn.Messagebus namespace:

ServiceClient sc = new ServiceClient();
sc.BaseUrl = "http://localhost:9080/call/";
var m1 = sc.CallService<TestMessage1>(new TestMessage1 { Id = 99 });
var m2 = sc.CallService<TestMessage1>(new TestMessage1 { Id = 100 });

This code calls the TestMessage1 service twice. The generic parameter to ServiceClient.CallService method is the expected return value type. It’s used for deserializing the Json response.

Other uses ¶

You are free to add your own ‘servlets’ to NGinn messagebus, you have just to inherit from the ServletBase class:

[UrlPattern(@"^/hello/")]
public class HelloWorldServlet : ServletBase
{
    public override void HandleRequest(IRequestContext ctx)
    {
        ctx.Output.WriteLine("Hello, world");
    }
}

This servlet will respond to http://localhost:9080/hello/ requests.
You also need to register your servlet during NGinn messagebus configuration:

MessageBusConfigurator.Begin()
                .SetConnectionStrings(dbConnectionStrings)
                .SetEndpoint(endpointName)
                .RegisterHttpHandlersFromAssembly(typeof(Program).Assembly)
(...)

RegisterHttpHandlersFromAssembly registers all servlet types that have an UrlPattern attribute.

What are sagas ¶

Saga is a piece of data that represents current state of some long-running process, a process that reacts to incoming messages. Sagas in nginn-messagebus are initiated by a message and then can process other incoming messages. Each saga is a distinct instance of a process and therefore handles only messages that are sent to that instance – nginn-messagebus makes sure saga messages are delivered to proper saga instances. Sagas are persistent and transactional.

But why are they a part of the message bus? ¶

Just for your convenience. Long running processes and async messaging quite often go together. Sagas make it easy to implement such processes so you don’t have to implement process state persistence, message correlation, concurrency, versioning and so on.

Example use of a saga – sending SMS messages ¶

Let’s take a real world example of a process that can be implemented using a saga: Our task is to create a service for sending SMS messages to customers. After the message is sent our service should also collect a delivery report from the SMS gateway and send a status update to the application that has sent the SMS. If no delivery report is received in 3 days we should inform the application that the SMS has not been delivered.

Sending an SMS in our example is a long running process because after sending the message to the SMS gateway we need to wait for the delivery report for that message. The wait time can be quite long – up to 3 days – so the process is certainly a long-running one.

Implementation ¶

Basically, our SMS service sits between an application and the SMS gateway and acts as an intermediary in communication between the application and the SMS gateway. The ‘value added’ by our service is handling of message delivery reports and a reliable API for sending SMS messages.

In our process of sending an SMS we’ll have to handle the following events:

1. Application sends an SMS (this initiates the process)
2. SMS service sends an SMS to the SMS gateway
3. SMS gateway sends us a delivery report
4. We (SMS service) send the delivery report to the application

5. Delivery time-out occurs (no delivery report received)

Each of these events will be represented by a message. Disclaimer: we wont’t go into details of how the communication with the SMS gateway is implemented – let’s just assume that proper messages are published to the messagebus by some component responsible for communication and that’s all.

So, let’s start with defining our messages:

/// <summary>
/// Application's request to send an SMS
/// </summary>
public class SendSMSRequest
{
    public string MSISDN { get; set; }
    public string Text { get; set; }
    public string MessageId { get; set; }
}

/// <summary>
/// SMS delivery report for the application
/// </summary>
public class SMSDeliveryReport
{
    public bool Delivered { get; set; }
    public string Reason { get; set; }
    public string MessageId { get; set; }
}

/// <summary>
/// Send a SMS to the SMS gateway
/// </summary>
public class SendSMSToGateway
{
    public string MSISDN { get; set; }
    public string Text { get; set; }
    public string SMSId { get; set; }
}

/// <summary>
/// SMS gateway's delivery status message
/// </summary>
public class GatewayDeliveryReport
{
    public bool Success { get; set; }
    public string SMSId { get; set; }
}

/// <summary>
/// Message for handling delivery report timeout
/// </summary>
public class SMSDeliveryTimeout
{
    public string MessageId { get; set; }
}

Now we’ll implement a saga that orchestrates all these messages. This is a saga class that doesn’t yet handle any messages but holds some state (a SmsState? object):

public class SmsSaga : Saga<SmsSaga.SmsState>
{
    public class SmsState
    {
        public DateTime MessageSentDate { get; set; }
    }
}

Now we’ll add message handlers to it. Let’s start with the SendSMSRequest message that initiates the whole process:

public class SmsSaga : Saga<SmsSaga.SmsState>, InitiatedBy<SendSMSRequest>
{
    public class SmsState
    {
        public DateTime MessageSentDate { get; set; }
        public bool DeliveryReportReceived { get; set; }
    }

    public IMessageBus MessageBus { get; set; }

    protected override void Configure()
    {
        //tell the message bus that saga ID will be passed in the MessageId field fo SendSMSRequest message
        SagaId<SendSMSRequest>(x => x.MessageId);
    }

    public void Handle(SendSMSRequest message)
    {
        //update saga state
        Data.MessageSentDate = DateTime.Now;
        Data.DeliveryReportReceived = false;

        //send a request to the SMS gateway
        MessageBus.NewMessage(new SendSMSToGateway { MSISDN = message.MSISDN, SMSId = message.MessageId, Text = message.Text })
            .Send("sql://gateway/Queue");

        //and schedule a timeout message to be delivered in 3 days from now
        MessageBus.NewMessage(new SMSDeliveryTimeout { MessageId = message.MessageId })
            .SetDeliveryDate(DateTime.Now.AddDays(3))
            .Publish();

    }
}

Lot of things are happening here. First of all, we implemented the InitiatedBy interface that adds the Handle(SendSMSRequest) method. This interface informs message bus that it should create a new instance of SmsSaga when a SendSMSRequest message arrives. Apart from that, we have implemented a Configure method. This method’s purpose is to define how messages handled by the saga should be used for identifying the saga instance. In our case we told message bus that the saga ID is in SendSMSRequest.MessageId field. Now we only have to take care that the MessageId is unique.

Now let’s take a look what SmsSaga is doing after receiving the SendSMSRequest. First, we update the saga state (in the Data property). Then we send a request to the SMS gateway (the SendSMSToGateway message). And finally, we publish the SMSDeliveryTimeout message to our local message queue and schedule it for delivery 3 days from now. That’s all – now we are waiting for the delivery report from the SMS gateway or the timeout – whichever occurs first. So, let’s add the implementation of these two cases:

public class SmsSaga : Saga<SmsSaga.SmsState>,
    InitiatedBy<SendSMSRequest>,
    IHandleSagaMessage<GatewayDeliveryReport>,
    IHandleSagaMessage<SMSDeliveryTimeout>
{
    public class SmsState
    {
        public DateTime MessageSentDate { get; set; }
        public bool DeliveryReportReceived { get; set; }
        public string SenderEndpoint { get; set; }
    }

    public IMessageBus MessageBus { get; set; }

    protected override void Configure()
    {
        //tell the message bus that saga ID will be passed in the MessageId field of SendSMSRequest message
        SagaId<SendSMSRequest>(x => x.MessageId);
        SagaId<GatewayDeliveryReport>(x => x.SMSId);
        SagaId<SMSDeliveryTimeout>(x => x.MessageId);
    }

    public void Handle(SendSMSRequest message)
    {
        //update saga state
        Data.MessageSentDate = DateTime.Now;
        Data.DeliveryReportReceived = false;
        Data.SenderEndpoint = MessageBus.CurrentMessageInfo.Sender; //store the sender's address for later use

        //send a request to the SMS gateway
        MessageBus.NewMessage(new SendSMSToGateway { MSISDN = message.MSISDN, SMSId = message.MessageId, Text = message.Text })
            .Send("sql://gateway/Queue");

        //and schedule a timeout message to be delivered in 3 days from now
        MessageBus.NewMessage(new SMSDeliveryTimeout { MessageId = message.MessageId })
            .SetDeliveryDate(DateTime.Now.AddDays(3))
            .Publish();

    }

    public void Handle(GatewayDeliveryReport message)
    {
        if (!Data.DeliveryReportReceived)
        {
            Data.DeliveryReportReceived = true;
            //send the report to the application
            MessageBus.NewMessage(new SMSDeliveryReport { Delivered = message.Success, Reason = message.Success ? "" : "Delivery failure", MessageId = Id })
                .Send(Data.SenderEndpoint);
        }
        //we could complete the saga here but let's wait for the timeout message
        //so it is processed without an error
    }

    public void Handle(SMSDeliveryTimeout message)
    {
        if (!Data.DeliveryReportReceived)
        {
            //inform the application about timeout
            MessageBus.NewMessage(new SMSDeliveryReport { Delivered = false, Reason = "Timeout", MessageId = this.Id })
                .Send(Data.SenderEndpoint);
        }
        SetCompleted(); //mark the saga as completed so it can be removed from the db
    }
}

Now the implementation is complete. We implemented two additional IHandleSagaMessage interfaces for GatewayDeliveryReport and SMSDeliveryTimeout messages. We also added saga ‘resolvers’ for these messages in the Configure method. Our saga is complete after the SMSDeliveryTimeout message is received. The handler of that message calls SetCompleted at the end to inform message bus that the saga is completed and can be deleted from the database (an no further messages are expected). We could also complete the saga after receiving the GatewayDeliveryReport message but then the saga would be deleted before receiving SMSDeliveryTimeout message and there would be an error reported for that message.

And basically, that’s all you need to know to implement sagas with nginn-messagebus.

Saga Id and correlation ¶

As you have seen in the application above, the Id assigned to the saga was retrieved from the initiating SendSMSRequest message and then it was present in all other message used in this scenario. This was the common identifier for all messages in that particular process instance and in this particular case the saga Id was the same as the message Id assigned by the application. The benefit was that everyone was using the same identifier so no mapping had to be done.

Sometimes, however, you will not specify saga Id in the initiating message – in this case nginn-messagebus will generate new, unique saga Id on receiving the message and this Id will be then used for correlating all other messages in that saga. This means that if the initiating application needs to ‘talk to’ the saga later it must know the saga’s Id. The id can be passed back in the reply to the initating message.

There’s one more way of correlating sagas and messages: NGinn-messagebus allows you to specify an additional identifier (‘CorrelationId?’) for any message sent:

MessageBus.NewMessage(new SomeMessage())
    .SetCorrelationId("298128993932320")
    .Publish();

If the message correlation id is specified and no other mappings have been defined in the saga’s Configure method, the message correlation Id will be used as the saga Id. It will also become the saga Id if specified for the initiating message.

Sagas and transactions ¶

Sagas are updated as a result of receiving a saga message and therefore all modifications are a part of the message receiving transaction. There’s one great thing about nginn-messagebus and sagas: if the saga state is kept in the same database as the message queues (the default setup) your saga handling code will not require a distributed transaction and this means a great performance.

CREATE TRIGGER On_Document_Update
   ON  Documents
   AFTER INSERT, UPDATE
AS
BEGIN
	INSERT INTO MQ_FeedNotify([from_endpoint]
           ,[to_endpoint]
           ,[subqueue]
           ,[insert_time]
           ,[retry_count]
           ,[retry_time]
           ,[msg_text])
    select
		'sql://db/MQ_FeedNotify',
		'sql://db/MQ_FeedNotify',
        'I',
        getdate(),
        0,
        getdate(),
        '{"$type":"MyApplication.DocumentUpdated,MyApplication", "DocumentId" : "' + convert(varchar, Id) + '" }'
	from Inserted
END
GO

This piece of SQL makes sure that every time a record is updated or added to Documents table our application will be notified with a message published to MQ_FeedNotify table.
It’s a very simple and elegant way (IMHO) to deliver notifications from SQL to a .Net application and it’s also very effective one – especially when you are handling batch updates. And you don’t have to worry about transactions or interfacing the outside world from sql server – everything is provided by SQL server and nginn-messagebus.

You are not limited to using triggers of course, message sending logic can be placed in a stored procedure or a plain sql statement. But SQL offers triggers for almost everything so it’s a natural way to detect some events and notify the outside world about them. And some triggers provide quite interesting information – like the Logon triggers in SQL Server 2008 that could be used for detecting database access.

Of course there are other ways of sending messages to a queue from SQL. You could implement your own queue table and C# code that reads from it, but nginn-messagebus does it already and has been verified in production. You could also send messages with SQL Message Broker – this is a message queuing mechanism built into SQL server. I don’t know it well enough to compare it to nginn-messagebus but with SMB you would also have to implement some C# code that delivers received messages to the application so again – consider using nginn-messagebus. Other options? Well, you could use some tricks like calling external scripts from SQL code or even embedding CLR code into SQL, but these look like dirty hacks compared to methods discussed above.

NGinn MessageBus provides publish-subscribe message distribution and supports transactional and durable messages only (as for now, because it’s still in development). All messages and subscriber information is persisted in SQL database and no other message transport mechanism is used. A message queue in NGinn MessageBus is simply a database table where messages are written and read in FIFO (more or less) manner. You might wonder why I decided to use SQL instead of well known messaging implementations like MSMQ, and here are the reasons:

• All application data in single database. As NGinn.BPM stores process state in SQL database it made a perfect sense to have all other data, including queued messages, in the same database. It simplifies maintaninance, backups and other management tasks.
• NGinn MessageBus provides ‘scheduled message’ functionality allowing you to schedule a message to be delivered at specified moment in future. This function is heavily used by NGinn BPM and there are lots of scheduled messages sitting in database and waiting to be delivered. It’s easy to handle this in a random-access database and very difficult to do it when you have a FIFO queue with sequential access only.
• The performance is good, in some situations much better than what I was able to get with MSMQ. For example, sending messages in a distributed transaction is faster in NGinn MessageBus than in MSMQ. NGinn BPM uses distributed transactions everywhere so this was important.

NGinn MessageBus is built around the concept of an Endpoint. Endpoint is an address to where messages can be delivered or from where they can be sent. Endpoint corresponds to a single message queue, with the same name (actually, the queue name is the endpoint). Usually you have one endpoint per process and in case of distributed applications each process has its own endpoint, and hence its own queue. If you have a single process you are using NGinn Message Bus in a ‘local’ mode, where messages are sent and received in the same queue. To extend it to distributed mode you have to set up several endpoints and provide message distribution rules by subscribing. One endpoint can subscribe at another endpoint for messages of particular type. After subscription is done all messages published at the publisher endpoint will be delivered to that endpoint locally and to all endpoints that have subscribed for particular message type.

NGinn Message Bus is built in a decentralized architecture – there is no central message router component, each pair of endpoints can communicate directly independent of any other endpoints.

From the application perspective you access NGinn MessageBus through the IMessageBus interface that provides functions for sending messages and also manages the process of receiving incoming messages and dispatching them to their handlers. NGinn Message Bus uses an IOC container to resolve message handlers so you have to register your message handling components in the container. Currently NGinn MessageBus project provides a configuration tool to set it up with the Castle Windsor container, but it’s possible to use other containers as well.

Some code examples

First of all, let’s see how to configure NGinn Message Bus with the Castle Windsor container. The source code contains a test project with very similar example:

Dictionary connStrings = new Dictionary();
connStrings["testdb1"] = "Data Source=(local);Initial Catalog=NGinn;User Id=nguser;Password=PASS";
connStrings["testdb2"] = "Data Source=(local);Initial Catalog=NGinn;User Id=nguser;Password=PASS";
IWindsorContainer wc1 = ConfigureMessageBus("sql://testdb1/MQueue1", connStrings);
IWindsorContainer wc2 = ConfigureMessageBus("sql://testdb2/MQueue2", connStrings);

IMessageBus mb1 = wc1.Resolve();
mb1.SubscribeAt("sql://testdb2/MQueue2", typeof(TestMessage1));

IMessageBus mb2 = wc2.Resolve();

The code above configures two instances of message bus component, for two different endpoints. Note that in order to configure a message bus endpoint we must specify the endpoint name in the form of sql://[connection alias]/[table name] and give the list of connection string aliases. Here we have two connection string aliases (testdb1 and testdb2), both pointing to the same database. So in the database we will have two message queue tables created, named MQueue1 and MQueue2 (NGinn MessageBus will create them automatically if the credentials provided in connection string allow for table creation).
There is also a call to mb1.SubscribeAt(“endpoint”, Message type) that tells message bus 1 (represented by sql://testdb1/MQueue1) to subscribe at endpoint sql://testdb2/MQueue2 to all messages of type TestMessage1.

And here is the ‘ConfigureMessageBus’ method code that performs actual configuration:

static IWindsorContainer ConfigureMessageBus(string endpointName, IDictionary dbConnectionStrings)
{
IWindsorContainer wc = Configure.Begin()
.SetConnectionStrings(dbConnectionStrings)
.SetEndpoint(endpointName)
.UseSqlSubscriptions()
.AddMessageHandler(typeof(TestHandler))
.BuildContainer();
return wc;

}

Now, after we have configured the message bus here is how to send some messages:

IMessageBus mb2 = wc2.Resolve();

using (System.Transactions.TransactionScope ts = new System.Transactions.TransactionScope())
{
for (int i = 0; i < 10; i++)
{
mb2.NewMessage(new TestMessage1("M" + i))
.SetDeliveryDate(DateTime.Now.AddSeconds(1 + i))
.SetLabel("Test XX" + i)
.Publish();
}
ts.Complete();
}

We open a transaction and then publish 10 messages. For each message we specify the delivery date (optional) and label (also optional). Messages are published to local endpoint, then they will be distributed according to currently configured subscriptions.

Finally, some code showing how to set up a message handler. Please take look above at the ConfigureMessageBus snippet. You’ll notice the line

.AddMessageHandler(typeof(TestHandler))

which registers a message handler component. This component looks like this:

public class TestHandler : IMessageConsumer {
private Logger log;

public TestHandler()
{
log = LogManager.GetLogger("TestHandler_" + this.GetHashCode());
}
#region IMessageConsumer Members

public void Handle(TestMessage1 message)
{
log.Info("TestMessage1 {0} ARRIVED", message.Id);
System.Threading.Thread.Sleep(1000);
}

#endregion

As you can see, message handler has to implement IMessageConsumerinterface which defines the ‘Handle’ method. Single component can implement the IMessageConsumer interface multiple times to be able to handle multiple types of messages.

And basically that’s it.