CODEWITHSUNDEEP
nservicebus
Yakaas
Yakaas

Reputation: 191

Recursive Bus.Send() with-in a Handler (Transactions, Threading, Tasks)

I have a handler similar to the following, which essentially responds to a command and sends a whole bunch of commands to a different queue.

    public void Handle(ISomeCommand message)
    {
        int i=0;
        while (i < 10000)
        {
            var command = Bus.CreateInstance<IAnotherCommand>();
            command.Id = i;
            Bus.Send("target.queue@d1555", command);
            i++;
        }
    }

The issue with this block is, until the loop is fully completed none of the messages appear in the target queue or in the outgoing queue. Can someone help me understand this behavior?

Also if I use Tasks to send messages within the Handler as below, messages appear immediately. So two questions on this,

  1. What's the explanation on Task based Sends to go through immediately?

  2. Are there are any ramifications on using Tasks with in message handlers?

    public void Handle(ISomeCommand message)
{
    int i=0;
    while (i < 10000)
    {
        System.Threading.ThreadPool.QueueUserWorkItem((args) =>
        {
            var command = Bus.CreateInstance<IAnotherCommand>();
            command.Id = i;
            Bus.Send("target.queue@d1555", command);
            i++;
        });
    }
}

Your time is much appreciated!

Upvotes: 3

Views: 1717

Answers (3)

Adam Fyles
Adam Fyles

Reputation: 6050

If you are trying to reduce overhead you can also bundle your messages. The signature for the send is Bus.Send(IMessage[]messages). If you can guarantee that you don't blow up the size limit for MSMQ, then you could Send() all the messages at once. If the size limit is an issue, then you can chunk them up or use the Databus.

Upvotes: 0

janovesk
janovesk

Reputation: 1138

First question: Picking a message from a queue, running all the registered message handlers for it AND any other transactional action(like writing new messages or writes against a database) is performed in ONE transaction. Either it all completes or none of it. So what you are seeing is the expected behaviour: picking a message from the queue, handling ISomeCommand and writing 10000 new IAnotherCommand is either done completely or none of it. To avoid this behaviour you can do one of the following:

  1. Configure your NServiceBus endpoint to not be transactional

    public class EndpointConfig : IConfigureThisEndpoint, AsA_Publisher,IWantCustomInitialization
{
    public void Init()
    {
        Configure.With()
            .DefaultBuilder()
            .XmlSerializer()
            .MsmqTransport()
            .IsTransactional(false)
            .UnicastBus();
    }
}

  2. Wrap the sending of IAnotherCommand in a transaction scope that suppresses the ambient transaction.

    public void Handle(ISomeCommand message)
{ 
    using (new TransactionScope(TransactionScopeOption.Suppress)) 
    { 
        int i=0; 
        while (i < 10000) 
        { 
            var command = Bus.CreateInstance(); 
            command.Id = i; 
            Bus.Send("target.queue@d1555", command); 
            i++; 
        } 
    } 
}

  3. Issue the Bus.Send on another thread, by either starting a new thread yourself, using System.Threading.ThreadPool.QueueUserWorkItem or the Task classes. This works because an ambient transaction is not automatically carried over to a new thread.

Second question: The ramifications of using Tasks, or any of the other methods I mentioned, is that you have no transactional quarantee for the whole thing.

How do you handle the case where you have generated 5000 IAnotherMessage and the power suddenly goes out?

If you use 2) or 3) the original ISomeMessage will not complete and will be retried automatically by NServiceBus when you start up the endpoint again. End result: 5000 + 10000 IAnotherCommands.

If you use 1) you will lose IAnotherMessage completely and end up with only 5000 IAnotherCommands.

Using the recommended transactional way, the initial 5000 IAnotherCommands would be discarded, the original ISomeMessage comes back on the queue and is retried when the endpoint starts up again. Net result: 10000 IAnotherCommands.

Upvotes: 4

Eben Roux
Eben Roux

Reputation: 13256

If memory serves NServiceBus wraps the calls to the message handlers in a TransactionScope if the transaction option is used and TransactionScope needs some help to be cross-thread friendly:

TransactionScope and multi-threading

Upvotes: 0

Related Questions