Batch Message Processing 3.0

Sometimes you might want to process a stream of incoming messages in batches rather than one at a time. This might be for performance reasons, or maybe there's some kind of business logic that makes more sense to calculate for batches, or maybe you want a logical "debounce" in how your system responds to the incoming messages.

INFO

The batching is supported both for messages published in process to local queues and from incoming messages from external transports.

Regardless, Wolverine has a mechanism to locally batch incoming messages and forward them to a batch handler. First, let's say that you have a message type called Item:

public record Item(string Name);

^{snippet source | anchor}

And for whatever reason, we need to process these messages in batches. To do that, we first need to have a message handler for an array of Item like so:

public static class ItemHandler
{
    public static void Handle(Item[] items)
    {
        // Handle this just like a normal message handler,
        // just that the message type is Item[]
    }
}

^{snippet source | anchor}

WARNING

At this point, Wolverine only supports an array of the message type for the batched handler

TIP

Batch message handlers are just like any other message handler and have no special rules about their capabilities

With that in our system, now we need to tell Wolverine to group Item messages, and we do that with the following syntax:

theHost = await Host.CreateDefaultBuilder()
    .UseWolverine(opts =>
    {
        opts.BatchMessagesOf<Item>(batching =>
        {
            // Really the maximum batch size
            batching.BatchSize = 500;
            
            // You can alternatively override the local queue
            // for the batch publishing. 
            batching.LocalExecutionQueueName = "items";

            // We can tell Wolverine to wait longer for incoming
            // messages before kicking out a batch if there
            // are fewer waiting messages than the maximum
            // batch size
            batching.TriggerTime = 1.Seconds();
            
        })
            
            // The object returned here is the local queue configuration that
            // will handle the batched messages. This may be useful for fine
            // tuning the behavior of the batch processing
            .Sequential();
    }).StartAsync();

^{snippet source | anchor}

And that's that! Just to bring this a little more into focus, here's an end to end test from the Wolverine codebase:

[Fact]
public async Task send_end_to_end_with_batch()
{
    // Items to publish
    var item1 = new Item("one");
    var item2 = new Item("two");
    var item3 = new Item("three");
    var item4 = new Item("four");

    Func<IMessageContext, Task> publish = async c =>
    {
        // I'm publishing the 4 items in sequence
        await c.PublishAsync(item1);
        await c.PublishAsync(item2);
        await c.PublishAsync(item3);
        await c.PublishAsync(item4);
    };

    // This is the "act" part of the test
    var session = await theHost.TrackActivity()
        
        // Wolverine testing helper to "wait" until
        // the tracking receives a message of Item[]
        .WaitForMessageToBeReceivedAt<Item[]>(theHost)
        .ExecuteAndWaitAsync(publish);

    // The four Item messages should be processed as a single 
    // batch message
    var items = session.Executed.SingleMessage<Item[]>();

    items.Length.ShouldBe(4);
    items.ShouldContain(item1);
    items.ShouldContain(item2);
    items.ShouldContain(item3);
    items.ShouldContain(item4);
}

^{snippet source | anchor}

Alright, with all that being said, here's a few more facts about the batch messaging support:

1. There is absolutely no need to create a specific message handler for the Item message, and in fact, you should not do so
2. The message batching is able to group the message batches by tenant id if your Wolverine system uses multi-tenancy

What about durable messaging ("inbox")?

The durable inbox behaves just a little bit differently for message batching. Wolverine will technically "handle" the individual messages, but does not mark them as handled in the message store until a batch message that refers to the original message is completely processed.