How to use Rebus in a web application

2015-08-21 by mookid8000 8 Comments

If you’re building web applications, you will often encounter situations where delegating work to some asynchronous background process can be beneficial. Let’s take a very simple example: Sending emails!

We’re in the cloud

Let’s say we’re hosting the website in Azure, as an Azure Web Site, and we need to send an email at the end of some web request. The most obvious way of doing that could be to new up an SmtpClient with our SendGrid credentials, and then just construct a MailMessage and send it.

While this solution is simple, it’s not good, because it makes it impossible for us to have higher uptime than SendGrid (or whichever email provider you have picked). In fact, every time we add some kind of synchronous call to the outside from our system, we impose their potential availability problems on ourselves.

We can do better 🙂

Let’s make it asynchronous

Now, instead of sending the email directly, let’s use Rebus in our website and delegate the integration to external systems, SendGrid included, to a message handler in the background.

This way, at the end of the web request, we just do this:

await _bus.Send(new SendEmail(recipient, subject, body));

1	await _bus.Send(new SendEmail(recipient, subject, body));

and then our work handling the web request is over. Now we just need to have something handle the SendEmail message.

Where to host the backend?

We could configure Rebus to use either Azure Service Bus or Azure Storage Queues to transport messages. If we do that, we can host the backend anywhere, including as a process running on a 386 with a 3G modem in the attic of an abandoned building somewhere, but I’ve got a way that’s even cooler: Just host it in the web site!

This way, we can have the backend be subject to the same auto-scaling and whatnot we might have configured for the web site, and if we’re a low traffic site, we can even get away with hosting it on the Free tier.

Moreover, our backend can be Git-deployed together with the website, which makes for a super-smooth experience.

How to do it?

It’s a good idea to consider the backend a separate application, even though we chose to deploy it as though it was one. This is just a simple example on how processes and applications are really orthogonal concepts – in general, it’s limiting to attempt to enforce a 1-to-1 between processes and applications(*).

What we should do is to have a 1-to-1 relationship between IoC container instances and applications, because that’s what IoC containers are for: To function as a container of one, logical application. In this case that means that we’ll spin up one Windsor container (or whichever IoC container is your favorite) for the web application, and one for the backend. In an OWIN Startup configuration class, it might look like this:

public class Startup
{
    public void Configuration(IAppBuilder app)
    {
        ConfigureWebApi(app);

        ConfigureBackend(app);
    }

    static void ConfigureWebApi(IAppBuilder app)
    {
        var httpConfiguration = new HttpConfiguration();
        WebApiConfig.Register(httpConfiguration);

        var webContainer = new WindsorContainer()
            .Install(new ApiHandlerInstaller())
            .Install(new WebRebusInstaller());

        httpConfiguration.UseWindsorContainer(webContainer);

        app.RegisterForDisposal(webContainer, "Windsor container for the web app");

        app.UseWebApi(httpConfiguration);
    }

    void ConfigureBackend(IAppBuilder app)
    {
        var backendContainer = new WindsorContainer()
            .Install(new RebusHandlerInstaller())
            .Install(new BackendRebusInstaller());

        app.RegisterForDisposal(backendContainer, "Windsor container for the background jobs");
    }
}

public class Startup

{

public void Configuration(IAppBuilder app)

{

ConfigureWebApi(app);

ConfigureBackend(app);

}

static void ConfigureWebApi(IAppBuilder app)

{

var httpConfiguration = new HttpConfiguration();

WebApiConfig.Register(httpConfiguration);

var webContainer = new WindsorContainer()

.Install(new ApiHandlerInstaller())

.Install(new WebRebusInstaller());

httpConfiguration.UseWindsorContainer(webContainer);

app.RegisterForDisposal(webContainer, "Windsor container for the web app");

app.UseWebApi(httpConfiguration);

}

void ConfigureBackend(IAppBuilder app)

{

var backendContainer = new WindsorContainer()

.Install(new RebusHandlerInstaller())

.Install(new BackendRebusInstaller());

app.RegisterForDisposal(backendContainer, "Windsor container for the background jobs");

}

In the code sample above, UseWindsorContainer and RegisterForDisposal are extension methods on IAppBuilder. UseWindsorContainer replaces Web API’s IHttpControllerActivator with a WindsorCompositionRoot like the one Mark Seemann wrote, and RegisterForDisposal looks like this:

public static void RegisterForDisposal(this IAppBuilder appBuilder, IDisposable disposable, string description = null)
{
    var descriptionOfDisposable = description ?? disposable.ToString();

    var context = new OwinContext(appBuilder.Properties);
    var token = context.Get<CancellationToken>("host.OnAppDisposing");
    if (token == CancellationToken.None)
    {
        Trace.TraceInformation("{0} could not be registered for disposal because the cancellation token could not be found", descriptionOfDisposable);
        return;
    }

    Trace.TraceInformation("Registering for disposal: {0}", descriptionOfDisposable);

    token.Register(() =>
    {
        Trace.TraceInformation("Disposing {0}", descriptionOfDisposable);

        disposable.Dispose();
    });
}

public static void RegisterForDisposal(this IAppBuilder appBuilder, IDisposable disposable, string description = null)

{

var descriptionOfDisposable = description ?? disposable.ToString();

var context = new OwinContext(appBuilder.Properties);

var token = context.Get<CancellationToken>("host.OnAppDisposing");

if (token == CancellationToken.None)

{

Trace.TraceInformation("{0} could not be registered for disposal because the cancellation token could not be found", descriptionOfDisposable);

return;

}

Trace.TraceInformation("Registering for disposal: {0}", descriptionOfDisposable);

token.Register(() =>

{

Trace.TraceInformation("Disposing {0}", descriptionOfDisposable);

disposable.Dispose();

});

}

which is how you make something be properly disposed when an OWIN-based application shuts down. Moreover, I’m using Windsor’s installer mechanism to register stuff in the containers.

Rebus configuration

Next thing to do, is to make sure that I configure Rebus correctly – since I have two separate applications, I will also treat them as such when I set up Rebus. This means that my web tier will have a one-way client, because it needs only to be able to bus.Send, whereas the backend will have a more full configuration.

The one-way client might be configured like this:

public class WebRebusInstaller : IWindsorInstaller
{
    public void Install(IWindsorContainer container, IConfigurationStore store)
    {
        var azureServiceBusConnectionString = ConfigurationManager
            .ConnectionStrings["azureServiceBusConnectionString"]
            .ConnectionString;

        Configure.With(new CastleWindsorContainerAdapter(container))
            .Transport(t => t.UseAzureStorageQueuesAsOneWayClient(azureServiceBusConnectionString))
            .Routing(r => r.TypeBased().MapAssemblyOf<SendEmail>("backend"))
            .Start();
    }
}

public class WebRebusInstaller : IWindsorInstaller

{

public void Install(IWindsorContainer container, IConfigurationStore store)

{

var azureServiceBusConnectionString = ConfigurationManager

.ConnectionStrings["azureServiceBusConnectionString"]

.ConnectionString;

Configure.With(new CastleWindsorContainerAdapter(container))

.Transport(t => t.UseAzureStorageQueuesAsOneWayClient(azureServiceBusConnectionString))

.Routing(r => r.TypeBased().MapAssemblyOf<SendEmail>("backend"))

.Start();

}

this registering an IBus instance in the container which is capable of sending SendEmail messages, which will be routed to the queue named backend.

And then, the backend might be configured like this:

public class BackendRebusInstaller : IWindsorInstaller
{
    public void Install(IWindsorContainer container, IConfigurationStore store)
    {
        var azureServiceBusConnectionString = ConfigurationManager
            .ConnectionStrings["azureServiceBusConnectionString"]
            .ConnectionString;

        Configure.With(new CastleWindsorContainerAdapter(container))
            .Transport(t => t.UseAzureStorageQueues(azureServiceBusConnectionString, "backend"))
            .Start();
    }
}

public class BackendRebusInstaller : IWindsorInstaller

{

public void Install(IWindsorContainer container, IConfigurationStore store)

{

var azureServiceBusConnectionString = ConfigurationManager

.ConnectionStrings["azureServiceBusConnectionString"]

.ConnectionString;

Configure.With(new CastleWindsorContainerAdapter(container))

.Transport(t => t.UseAzureStorageQueues(azureServiceBusConnectionString, "backend"))

.Start();

}

Only thing left is to write the SendEmailHandler:

Email message handler

public class SendEmailHandler : IHandleMessages<SendEmail>
{
    public async Task Handle(SendEmail message)
    {
        var mail = CreateMailMessage(message);
        var client = GetSmtpClient();

        using(client)
        {
            await client.SendMailAsync(mail);
        }
    }

    MailMessage CreateMailMessage(SendEmail message)
    {
         //...
    }

    SmtpClient GetSmtpClient()
    {
         //...
    }
}

public class SendEmailHandler : IHandleMessages<SendEmail>

{

public async Task Handle(SendEmail message)

{

var mail = CreateMailMessage(message);

var client = GetSmtpClient();

using(client)

{

await client.SendMailAsync(mail);

}

MailMessage CreateMailMessage(SendEmail message)

{

//...

}

SmtpClient GetSmtpClient()

{

//...

}

Conclusion

Hosting a Rebus endpoint inside an Azure Web Site can be compelling for several reasons, where smooth deployment of cohesive units of web+backend can be made trivial.

I’ve done this several times myself, in web sites with multiple processes, including sagas and timeouts stored in SQL Azure, and basically everything Rebus can do – and so far, it has worked flawlessly.

Depending on your requirements, you might need to flick on the “Always On” setting in the portal

Skærmbillede 2015-08-21 kl. 11.44.22

so that the site keeps running even though it’s not serving web requests.

Final words

If anyone has experiences doing something similar to this in Azure or with another cloud service provider, I’d be happy to hear about it 🙂

(*) This 1-to-1-ness is, in my opinion, a thing that the microservices community does nok mention enough. I like to think about processes and applications much the same way as Udi Dahan describes it.

Another Rebus extension example

2015-08-13 by mookid8000 Leave a comment

In the previous post I showed how Rebus’ subscription storage could report itself as “centralized”, which would provide a couple of benefits regarding configuration. In the post before the previous post, I showed how Rebus could be extended in order to execute message handlers inside a System.Transactions.TransactionScope, which was easy to do, without touching a single bit in Rebus core.

This time, I’ll combine the stuff I talked about in the two posts, and show how the Azure Service Bus transport can hook itself into the right places in order to function as a multicast-enabled transport, i.e. a transport that natively supports publish/subscribe and thus can relieve Rebus of some of its work.

Again, let’s take it from the outside and in – I would love it if Rebus could be configured to use Azure Service Bus simply by doing something like this:

var activator = new BuiltinHandlerActivator();

Configure.With(activator)
    .Transport(t => t.UseAzureServiceBus(connectionString, "my_input_queue"))
    .Start();

var activator = new BuiltinHandlerActivator();

Configure.With(activator)

.Transport(t => t.UseAzureServiceBus(connectionString, "my_input_queue"))

.Start();

to configure each endpoint, and then await bus.Subscribe<SomeMessage>() and await bus.Publish(new SomeMessage("woohoo it works!!")) in order to subscribe to messages and publish them, and then just sit back and see published messages flow to their subscribers without any additional work.

Now, what would it require to make that work?

Subscriptions must be stored somewhere

No matter which kind of technology you use to move messages around, and no matter how complex logic they support, I bet they somehow build on some kind of message queue building block – i.e. a thing, into which you may put messages meant for some specific recipient to receive, and out of which one specific recipient can get its messages – and then everything that the transport can do with the messages, like routing, filtering, fan-out, etc., is implemented as logic that hooks into places and does stuff, but it will always end out with messages going into message queues.

Since Rebus is made to be able to run right on top of some pretty basic queueing systems, like e.g. MSMQ, and then has some of its functions going on in “user space” (like pub/sub messaging), it has an abstraction for something that persists subscriptions: ISubscriptionStorage – this is the way a publisher “remembers” which queues to send to when it publishes a message.

When a transport offers its own pub/sub mechanism (and Azure Service Bus does that via topics and subscriptions) it means that it effectively works as a centralized implementation of ISubscriptionStorage (as described in the previous post) – and in fact, it turns out that that is the proper place to hook in in order to take advantage of the native pub/sub mechanism.

Let’s do it 🙂

How to replace the subscription storage

Similar to what I showed in the previous Rebus extension example, the UseAzureServiceBus function above is an extension method – it uses Injectionist to register an instance of the transport, and then it sets up resolvers to use the transport as the primary ITransport and ISubscriptionStorage implementations – it looks like this:

public static AzureServiceBusTransportSettings UseAzureServiceBus(this StandardConfigurer<ITransport> configurer, string connectionStringNameOrConnectionString, string inputQueueAddress)
{
    var connectionString = GetConnectionString(connectionStringNameOrConnectionString);
    var settingsBuilder = new AzureServiceBusTransportSettings();

    // register instance that implements ITransport and ISubscriptionStorage
    configurer
        .OtherService<AzureServiceBusTransport>()
        .Register(c =>
        {
            var transport = new AzureServiceBusTransport(connectionString, inputQueueAddress);

            if (settingsBuilder.PrefetchingEnabled)
            {
                transport.PrefetchMessages(settingsBuilder.NumberOfMessagesToPrefetch);
            }

            if (settingsBuilder.AutomaticPeekLockRenewalEnabled)
            {
                transport.AutomaticallyRenewPeekLock();
            }

            return transport;
        });

    // resolve ISubscriptionStorage by forwarding to the transport instance
    configurer
        .OtherService<ISubscriptionStorage>()
        .Register(c => c.Get<AzureServiceBusTransport>());

    // resolve ITransport by forwarding to the transport instance
    configurer.Register(c => c.Get<AzureServiceBusTransport>());

    return settingsBuilder;
}

public static AzureServiceBusTransportSettings UseAzureServiceBus(this StandardConfigurer<ITransport> configurer, string connectionStringNameOrConnectionString, string inputQueueAddress)

{

var connectionString = GetConnectionString(connectionStringNameOrConnectionString);

var settingsBuilder = new AzureServiceBusTransportSettings();

// register instance that implements ITransport and ISubscriptionStorage

configurer

.OtherService<AzureServiceBusTransport>()

.Register(c =>

{

var transport = new AzureServiceBusTransport(connectionString, inputQueueAddress);

if (settingsBuilder.PrefetchingEnabled)

{

transport.PrefetchMessages(settingsBuilder.NumberOfMessagesToPrefetch);

}

if (settingsBuilder.AutomaticPeekLockRenewalEnabled)

{

transport.AutomaticallyRenewPeekLock();

}

return transport;

});

// resolve ISubscriptionStorage by forwarding to the transport instance

configurer

.OtherService<ISubscriptionStorage>()

.Register(c => c.Get<AzureServiceBusTransport>());

// resolve ITransport by forwarding to the transport instance

configurer.Register(c => c.Get<AzureServiceBusTransport>());

return settingsBuilder;

}

Now, the Azure Service Bus transport just needs to perform some meaningful actions as the subscription storage it is now claiming to be. Let’s take a look at

How to be a subscription storage

Subscription storages need to implement this interface:

public interface ISubscriptionStorage
{
    Task<string[]> GetSubscriberAddresses(string topic);

    Task RegisterSubscriber(string topic, string subscriberAddress);

    Task UnregisterSubscriber(string topic, string subscriberAddress);

    bool IsCentralized { get; }
}

public interface ISubscriptionStorage

{

Task<string[]> GetSubscriberAddresses(string topic);

Task RegisterSubscriber(string topic, string subscriberAddress);

Task UnregisterSubscriber(string topic, string subscriberAddress);

bool IsCentralized { get; }

}

where you already know that the IsCentralized property must return true, indicating that subscribers can register themselves directly. And then, because it’s Azure Service Bus, we just need to

ensure the given topic exists, and
create a subscription on the given topic with ForwardTo set to the subscriber’s input queue

in order to start receiving the subscribed-to events. And that is in fact what the Azure Service Bus transport is doing now 🙂

Only thing left for this to work, is this:

Rebus must publish in the right way

When Rebus publishes a message, it goes through the following sequence:

Asks the subscription storage for subscribers of the given topic
For each subscriber: Asks the transport to send the message to that subscriber

Now, since the Azure Service Bus transport is both subscription storage and transport, we’ll take advantage of this fact by having GetSubscriberAddresses return only one single – fake! – subscriber address, on the form subscribers/<topic>(*).

And then, when the transport detects a destination address starting with the subscribers/ prefix (which cannot be a valid Azure Service Bus queue name), the transport will use a TopicClient for the right topic to publish the message instead of the usual QueueClient.

Conclusion

Rebus (since 0.90.8) can take advantage of Azure Service Bus’ native support for publish/subscribe, which means that you need not worry about routing of events or configuring any other kind of subscription storage.

(*) Azure Service Bus does not support “,”, “+”, and other characters in topics, and these characters can often be found in .NET type names (which Rebus likes to use as topics), Azure Service Bus transport will normalize topics by removing these illegal characters. In fact, all non-digit non-letter characters will be replaced by “_”, causing "System.String, mscorlib" to become a topic named "system_string__mscorlib".

Storing subscriptions

2015-08-10 by mookid8000 Leave a comment

TL;DR: With Rebus2, subscription storages can be configured to be “centralized”, which means that all endpoints are allowed to register themselves as subscribers directly in the subscription storage. This way, since the central subscription storage is shared between all endpoints, no additional routing information is needed in order to implement pub/sub.

Long version:

Rebus has always had a subscription storage, which was the persistence abstration that a publisher would use to remember who subscribed to what.

Basic workings of the subscription mechanism

In the most basic scenario, it works like this:

1: Subscriber locates publisher

A subscriber, some_subscriber wants to subscribe to all string events.

In order to do this, the subscriber asks The Router (I’ll talk some more about The Router in another blog post, I promise 🙂 ): “Router – who owns System.String, mscorlib?” to which The Router replies: “ some_publisher” (The Router is just concise like that…)

2: Subscriber subscribes

The subscriber now sends a SubscribeRequest to some_publisher, which is basically saying something like “hi, I’m some_subscriber – I’d like to subscribe to System.String, mscorlib.

3: Publisher remembers

Having received the SubscribeRequest, the publisher then saves this information to its subscription storage (e.g. a table in SQL Server) in the form of this simple tuple: ("System.String, mscorlib", "some_subscriber").

4: Publisher publishes

For all eternity (or until the subscriber unsubscribes again), the publisher will then publish its string events by checking its subscription storage for subscribers of System.String, mscorlib.

Let’s pretend that it gets these two bad boys:

("System.String, mscorlib", "some_subscriber")
("System.String, mscorlib", "another_subscriber")

With these two in hand, the publisher will then just go on and send the event to each subscriber directly, i.e. to some_subscriber and another_subscriber in this case.

To sum it up

The basic pub/sub sequence can be sketched as this, which more accurately could be called “sub/pub”, because that’s the order of events – first, steps 1, 2, and 3 (router lookup omitted for clarity)

and then step 4

pub

A nifty trick with new Rebus

Check out the current layout of the ISubscriptionStorage interface:

public interface ISubscriptionStorage
{
    Task<string[]> GetSubscriberAddresses(string topic);

    Task RegisterSubscriber(string topic, string subscriberAddress);

    Task UnregisterSubscriber(string topic, string subscriberAddress);

    bool IsCentralized { get; }
}

public interface ISubscriptionStorage

{

Task<string[]> GetSubscriberAddresses(string topic);

Task RegisterSubscriber(string topic, string subscriberAddress);

Task UnregisterSubscriber(string topic, string subscriberAddress);

bool IsCentralized { get; }

}

In Rebus2 (i.e. versions 0.90.* and on), the subscription storage has had the ability to be “centralized” – i.e. an implementation of Rebus’ ISubscriptionStorage could return true from the IsCentralized property, which will cause Rebus to bypass the publisher lookup in the router and the sending of the SubscribeRequest, thereby shortcutting the subscription process a great deal.

Which makes perfect sense, because it doesn’t matter who makes the ("System.String, mscorlib", "some_subscriber") registration in the subscription storage, if only the publisher gets it when it’s time to publish some messages. Therefore – if a subscription storage is configured to be centralized, the subscription process is as short as this:

1: Subscriber subscribes

The subscriber saves its subscription to the subscription storage (e.g. a table in SQL Server) in the form of this simple tuple: ("System.String, mscorlib", "some_subscriber").

and then we would already be ready for

4: Publisher publishes

(which I will not repeat here)

To sum it up

When the subscription storage is centralized, the subscription process can be sketched like this:

subscribe2

When to use what?

The distributed way

Pros: The basic IsCentralized = false way is the most distributed form of publish/subscribe, since all subscribers rely on knowing the addresses of their dependent publishers only, and need not worry about having a connection to a subscription database somewhere.

With the non-centralized solution, each publisher can have its own subscription storage, which can e.g. be a local SQL Server Express, a JSON file, or whatever each publisher feels like. This makes the non-centralized way more resilient to failures, because there’s no central database required for publishers to work.

Cons: With the non-centralized solution, all subscribers must have an endpoint mapping in their routers, mapping one single publisher as the owner of each event type. More discipline is required during development, and there’s simply more moving parts at runtime.

The centralized way

Pros: The new IsCentralized = true way is easier to get started with, and it may be easier to grasp for newcomers. It also alleviates all subscribers of the burden of having to map publishers as owners of all the different types of events.

Cons: All publishers and subscribers are required to have the centralized subscription storage configured, and thus everyone needs to be able to connect to the same central database.

A Rebus extension example

2015-08-05 by mookid8000 Leave a comment

In my previous post about the ready-made pipeline step injector, I showed how to easily inject a new pipeline step into the pipeline at a position relative to another step.

A concrete example of doing this could be the Rebus.TransactionScope extension, which wraps the execution of handlers in a .NET System.Transactions.TransactionScope.

Let’s take this one from the outside-in. I want a configuration experience that looks like this:

Configure.With(activator)
    .Transport(t => t.UseInMemoryTransport(new InMemNetwork(), "txtest"))
    .Options(o =>
    {
        o.HandleMessagesInsideTransactionScope();
    })
    .Start();

Configure.With(activator)

.Transport(t => t.UseInMemoryTransport(new InMemNetwork(), "txtest"))

.Options(o =>

{

o.HandleMessagesInsideTransactionScope();

})

.Start();

In the example shown, HandleMessagesInsideTransactionScope is an extension method on Rebus’ OptionsConfigurer.

You know about extension methods? They’re great, they allow you to bring in a NuGet package from the outside, which contributes with a function that magically pop up somewhere else, e.g. on a fluent configuration builder thingie – exactly what we need 🙂

From now on, since you’ve read the post about the PipelineStepInjector, it’s so simple that you can probably already imagine what I’m about to show you – here’s the implementation:

public static class TransactionScopeConfigurationExtensions
{
    public static OptionsConfigurer HandleMessagesInsideTransactionScope(this OptionsConfigurer configurer)
    {
        configurer.Decorate<IPipeline>(c =>
        {
            var pipeline = c.Get<IPipeline>();
            var stepToInject = new TransactionScopeIncomingStep();

            return new PipelineStepInjector(pipeline)
                .OnReceive(stepToInject, PipelineRelativePosition.Before, typeof (DispatchIncomingMessageStep));
        });

        return configurer;
    }
}

public static class TransactionScopeConfigurationExtensions

{

public static OptionsConfigurer HandleMessagesInsideTransactionScope(this OptionsConfigurer configurer)

{

configurer.Decorate<IPipeline>(c =>

{

var pipeline = c.Get<IPipeline>();

var stepToInject = new TransactionScopeIncomingStep();

return new PipelineStepInjector(pipeline)

.OnReceive(stepToInject, PipelineRelativePosition.Before, typeof (DispatchIncomingMessageStep));

});

return configurer;

}

As you can see, this is a fairly declarative way of positioning our TransactionScopeIncomingStep right before the message is dispatched (in DispatchIncomingMessageStep).

Only thing left is to show the actual implementation of the step:

class TransactionScopeIncomingStep : IIncomingStep
{
    public async Task Process(IncomingStepContext context, Func<Task> next)
    {
        using (var scope = new System.Transactions.TransactionScope(TransactionScopeAsyncFlowOption.Enabled))
        {
            await next();
            scope.Complete();
        }
    }
}

class TransactionScopeIncomingStep : IIncomingStep

{

public async Task Process(IncomingStepContext context, Func<Task> next)

{

using (var scope = new System.Transactions.TransactionScope(TransactionScopeAsyncFlowOption.Enabled))

{

await next();

scope.Complete();

}

How about that? 😀

Please note that the Rebus.TransactionScope extension requires at least .NET 4.5.1, because that’s the version of the framework that introduced transaction scopes that would flow to continuations properly when using async/ await.

One way of extending the Rebus pipeline

2015-08-02 by mookid8000 Leave a comment

New Rebus processes messages through two pipelines: one for ougoing messages and one for incoming messages.

When you configure the bus, you can log the contents of the two pipelines (including their documentation if they have been decorated with the [StepDocumentation("bla bla bla")] attribute) by calling .Options(o => o.LogPipeline(verbose:true|false)) in the configuration, like so:

Configure.With(new BuiltinHandlerActivator())
    .Transport(t => t.UseInMemoryTransport(new InMemNetwork(), "test"))
    .Options(o => o.LogPipeline(verbose:true))
    .Start();

Configure.With(new BuiltinHandlerActivator())

.Transport(t => t.UseInMemoryTransport(new InMemNetwork(), "test"))

.Options(o => o.LogPipeline(verbose:true))

.Start();

which will output something like this to whichever logger is configured:

------------------------------------------------------------------------------
Message pipelines
------------------------------------------------------------------------------
Send pipeline:
    Rebus.Pipeline.Send.AssignGuidMessageIdStep
        Assigns a new GUID as the 'rbs2-msg-id' header if the message
        being sent does not already have a message ID.

    Rebus.Pipeline.Send.AssignReturnAddressStep
        If the outgoing message does not already have a 'rbs2-return-address'
        header, this step will assign the input queue address
        of the sending endpoint as the return address.

    Rebus.Pipeline.Send.AssignDateTimeOffsetHeader
        Sets the 'rbs2-senttime' header of the outgoing message
        to the current local time as a DateTimeOffset serialized
        with the 'O' format string.

    Rebus.Pipeline.Send.FlowCorrelationIdStep
        Sets the 'rbs2-corr-id' header of the outgoing message
        to one of the following three things:
        
        1) The correlation ID of the message currently being handled.
        2) The message ID of the message currently being handled.
        3) The message´s own message ID.

    Rebus.Pipeline.Send.SerializeOutgoingMessageStep
        Serializes the outgoing message using the configured
        serializer, storing the resulting transport message back
        to the context.

    Rebus.Pipeline.Send.SendOutgoingMessageStep
        Final step that uses the current transport to send the transport
        message found in the context to all addresses found by looking
        up the DestinationAddress object from the context.


Receive pipeline:
    Rebus.Retry.Simple.SimpleRetryStrategyStep
        Wraps the invocation of the entire receive pipeline in an exception
        handler, tracking the number of times the received message
        has been attempted to be delivered.
        
        If the maximum number of delivery attempts is reached,
        the message is moved to the error queue.

    Rebus.Pipeline.Receive.HandleDeferredMessagesStep
        If the incoming message should not be handled now, this step
        saves the message until it is time to deliver the message.
        
        This is done by checking if the incoming message has a 'rbs2-deferred-until'
        header with a desired time to be delivered.

    Rebus.Pipeline.Receive.DeserializeIncomingMessageStep
        Deserializes the current transport message using the configured
        serializer, saving the deserialized message back to the context.

    Rebus.Pipeline.Receive.ActivateHandlersStep
        Looks at the incoming message and decides how to handle
        it. A HandlerInvokers object is saved to the context to be
        invoked later.

    Rebus.Pipeline.Receive.LoadSagaDataStep
        Looks at the handler invokers in the context and sees if there´s
        one or more saga handlers in there.
        
        If that´s the case, relevant saga data is loaded/created,
        and the rest of the pipeline gets invoked.
        
        Afterwards, all the created/loaded saga data is updated
        appropriately.

    Rebus.Pipeline.Receive.DispatchIncomingMessageStep
        Gets all the handler invokers from the current context
        and invokes them in order.
        
        Please note that each invoker might choose to ignore the invocation
        internally.
        
        If no invokers were found, a RebusApplicationException
        is thrown.

------------------------------------------------------------------------------

------------------------------------------------------------------------------

Message pipelines

------------------------------------------------------------------------------

Send pipeline:

Rebus.Pipeline.Send.AssignGuidMessageIdStep

Assigns a new GUID as the 'rbs2-msg-id' header if the message

being sent does not already have a message ID.

Rebus.Pipeline.Send.AssignReturnAddressStep

If the outgoing message does not already have a 'rbs2-return-address'

header, this step will assign the input queue address

of the sending endpoint as the return address.

Rebus.Pipeline.Send.AssignDateTimeOffsetHeader

Sets the 'rbs2-senttime' header of the outgoing message

to the current local time as a DateTimeOffset serialized

with the 'O' format string.

Rebus.Pipeline.Send.FlowCorrelationIdStep

Sets the 'rbs2-corr-id' header of the outgoing message

to one of the following three things:

1) The correlation ID of the message currently being handled.

2) The message ID of the message currently being handled.

3) The message´s own message ID.

Rebus.Pipeline.Send.SerializeOutgoingMessageStep

Serializes the outgoing message using the configured

serializer, storing the resulting transport message back

to the context.

Rebus.Pipeline.Send.SendOutgoingMessageStep

Final step that uses the current transport to send the transport

message found in the context to all addresses found by looking

up the DestinationAddress object from the context.

Receive pipeline:

Rebus.Retry.Simple.SimpleRetryStrategyStep

Wraps the invocation of the entire receive pipeline in an exception

handler, tracking the number of times the received message

has been attempted to be delivered.

If the maximum number of delivery attempts is reached,

the message is moved to the error queue.

Rebus.Pipeline.Receive.HandleDeferredMessagesStep

If the incoming message should not be handled now, this step

saves the message until it is time to deliver the message.

This is done by checking if the incoming message has a 'rbs2-deferred-until'

header with a desired time to be delivered.

Rebus.Pipeline.Receive.DeserializeIncomingMessageStep

Deserializes the current transport message using the configured

serializer, saving the deserialized message back to the context.

Rebus.Pipeline.Receive.ActivateHandlersStep

Looks at the incoming message and decides how to handle

it. A HandlerInvokers object is saved to the context to be

invoked later.

Rebus.Pipeline.Receive.LoadSagaDataStep

Looks at the handler invokers in the context and sees if there´s

one or more saga handlers in there.

If that´s the case, relevant saga data is loaded/created,

and the rest of the pipeline gets invoked.

Afterwards, all the created/loaded saga data is updated

appropriately.

Rebus.Pipeline.Receive.DispatchIncomingMessageStep

Gets all the handler invokers from the current context

and invokes them in order.

Please note that each invoker might choose to ignore the invocation

internally.

If no invokers were found, a RebusApplicationException

is thrown.

------------------------------------------------------------------------------

As you can see, it’s fairly easy to get an understanding of what exactly is going on with these messages going in and out 🙂

When the bus needs either of the pipelines, it will use the IPipeline service to get them – IPipeline is pretty simple, it looks like this:

public interface IPipeline
{
    IEnumerable<IOutgoingStep> SendPipeline();
    IEnumerable<IIncomingStep> ReceivePipeline();
}

public interface IPipeline

{

IEnumerable<IOutgoingStep> SendPipeline();

IEnumerable<IIncomingStep> ReceivePipeline();

}

Now, if you’ve read my previous post about how Rebus uses Injectionist to satisfy its dependencies, you can probably envision by now a decorator-based approach to modify the pipeline(s) in order to implement new functionality… and you are absolutely right, that is exactly what I’m about to show you 🙂

In fact, this scenario where you want to use a decorator to inject something into either of the pipelines is so common that it has a nifty decorator prepared to cover most scenarios: PipelineStepInjector. The PipelineStepInjector is a IPipeline decorator that can be configured to inject new steps into either pipeline, at a position relative to another step.

For example, if I have created a new implementation of IIncomingStep, LogResolvedMessageHandlersStep, that logs the type names of resolved message handlers, a natural place to inject that step would be right after ActivateHandlersStep, whose responsitiblity it is to resolve all implementations compatible with IHandleMessages<SomeMessage> from the configured handler activator.

In order to inject this new step, I could do it like this:

injectionist.Register<IPipeline>(c => {
    var pipeline = c.Get<IPipeline>();
    var stepToInject = new LogResolvedMessageHandlersStep();

    return new PipelineStepInjector(pipeline)
        .OnReceive(stepToInject, PipelineRelativePosition.After, typeof(ActivateHandlersStep));
}, isDecorator: true);

injectionist.Register<IPipeline>(c => {

var pipeline = c.Get<IPipeline>();

var stepToInject = new LogResolvedMessageHandlersStep();

return new PipelineStepInjector(pipeline)

.OnReceive(stepToInject, PipelineRelativePosition.After, typeof(ActivateHandlersStep));

}, isDecorator: true);

which will cause my LogResolvedMessageHandlersStep to be positioned right after ActivateHandlersStep.

Nitfy, huh? 😀

Please note that PipelineStepInjector is not meant to be able to cover all scenarios. I could easily come up with complex injection rules that could not readily be handled simply by positioning each step relative to one of the existing steps, but so far it has covered quite a few scenarios nicely.