July 2015 – mookid on code

The new Rebus (i.e. “Rebus 2”, which is currently available from version 0.90.* on NuGet) is very easy to extend, due to the fact that the configuration API is backed by a very simple IoC-like container: the “Injectionist“.

The Injectionist is based on a simple general principle: For each type of service that you register (e.g. ISubscriptionStorage), you are allowed to register ONE primary implementation of that service (e.g. SqlServerSubscriptionStorage), and any number of decorators to that service (e.g. CachingSubscriptionStorage, etc.).

All registrations in the Injectionist are functions that must return objects – so for example to register our subscription storage example from above, we would do it like this:

injectionist.Register<ISubscriptionStorage>(c => {
    return new SqlServerSubscriptionStorage(connectionString, tableName);
});

injectionist.Register<ISubscriptionStorage>(c => {
    var subscriptionStorage = c.Get<ISubscriptionStorage>();

    return new CachingSubscriptionStorage(subscriptionStorage);
}), isDecorator: true);

injectionist.Register<ISubscriptionStorage>(c => {

return new SqlServerSubscriptionStorage(connectionString, tableName);

});

injectionist.Register<ISubscriptionStorage>(c => {

var subscriptionStorage = c.Get<ISubscriptionStorage>();

return new CachingSubscriptionStorage(subscriptionStorage);

}), isDecorator: true);

and then, since the RebusBus instance is registered like this:

injectionist.Register<IBus>(c => new RebusBus(c.Get<ISubscriptionStorage>(), ...));

1	injectionist.Register<IBus>(c => new RebusBus(c.Get<ISubscriptionStorage>(), ...));

the configuration API can end up calling injectionist.Get<IBus>() in order to build a bus instance with the chosen implementations of the various abstractions.

In addition to making it easy to gradually build a configuration, the Injectionist enables YOU to intercept almost anything by registering your own decorators…

Want to mutate incoming/outgoing messages? There’s several ways of doing that, but one way would be to register a decorator for the ISerializer interface.

Want to measure the time it takes to dispatch a message? Again, there’s several ways to do that, but one way could be to decorate the IPipelineInvoker interface.

And so on 🙂

I’ve been meaning to do this for a while, and while I am aware of the danger of doing it, I’ve done it anyways: I’ve rewritten Rebus!

Why would you do that?

Rebus was initially made to make me happy. I made it as an implementation of the parts of NServiceBus that I liked best in order to use it on a couple of projects, and thus I did not put too much effort into making a long-lasting design. Also, I aligned all APIs very closely to their NServiceBus counterparts, although I did loosen up on this as time went by.

In addition to this, Rebus was my first open-source project, and to this date, more than 20 besides me have contributed code to it, and many more are using it. It’s been fun to manage a project with actual contributors, and at times it’s been challenging to govern the project with regards to features, quality, ease-of-use, responsiveness, etc.

To sum it up: I have learned a lot since I started the Rebus project!

And usually, when you learn things, you want to put those things to use. At this point, I feel I’ve learned so much that I needed to put those things back into Rebus,

Why would you do it like that?

Rebus has – since its inception almost 5 years ago – done a good job of keeping message flowing and developers happy, so why would I rewrite it from scratch?

I’ve been meaning to improve a lot of things in Rebus for quite a while now, like reducing complexity, introduce message pipelines for incoming and outgoing messages, etc. I could probably have undertaken this as an incremental task, but my feeling is that the end result would not be nearly as clean as it will be with a rewrite.

Therefore, my conclusion was this: Rebus is small enough that a rewrite was the best way to refactor it (given the changes I wanted to make, obviously)!

Also, I did not actually throw old Rebus away. Rebus 2 is async to the core, and the OWIN-pipeliney design is heavily inspired by my (very brief) work with the NServiceBus team, but I kept everything aligned with old Rebus as far as it made sense, thus allowing me to port most of the integration implementations forward with very few modifications.

What has changed?

Rebus 2 is async to the core. All APIs are, in their most basic incarnation, async. Therefore, you await bus.Send(yourMessages) now. In other words, one single physical thread can do a lot of work, while consuming almost no memory, and almost no CPU.
Rebus has pipelines. Incoming and outgoing messages are passed through pipelines, which allows for extreme flexibility in how Rebus behaves. Several of Rebus’ “built-in” functionalities like e.g. compression and encryption are simply extra pipeline steps that are applied to incoming and outgoing messages.
Routing is based on topics. Topics based on .NET type names (which is how old Rebus worked) is thus just one case of a more general scheme where any string can be a topic.
There’s no batching. Batching (i.e. the fact that each and every transport message was actually a container of zero or more logical messages) provided almost no benefit, considering the amount of code it took to implement it. You can easily implement batching yourself in those few cases where you want it.
There’s an in-memory transport, an Amazon SQS transport, an Azure Storage Queues transport, a Jil serializer, and more cool stuff to come!
Idempotent sagas can be had by deriving your handler off of IdempotentSaga instead of Saga.
…and much more 🙂

Show me the code!

Chances are that many Rebus users will not experience too many differences. Check this out – here’s how I start the usual MSMQ-enabled endpoint, capable of receiving requests and delivering replies:

// usually, you'd use a real IoC container here
var activator = new BuiltinHandlerActivator();

activator.Handle<SomeRequest>(async (bus, request) => {
  DoStuffWith(request);
  await bus.Reply(new SomeReply("thanks, dude!"));
});

// start the bus
var bus = Configure.With(activator)
    .Transport(t => t.UseMsmq("myInputQueue"))
    .Start();

// run until we quit
using(bus) {
    Console.WriteLine("Press ENTER to quit");
    Console.ReadLine();
}

// usually, you'd use a real IoC container here

var activator = new BuiltinHandlerActivator();

activator.Handle<SomeRequest>(async (bus, request) => {

DoStuffWith(request);

await bus.Reply(new SomeReply("thanks, dude!"));

});

// start the bus

var bus = Configure.With(activator)

.Transport(t => t.UseMsmq("myInputQueue"))

.Start();

// run until we quit

using(bus) {

Console.WriteLine("Press ENTER to quit");

Console.ReadLine();

}

and then, in order to send SomeMessage to the endpoint, do this:

await bus.Send(new SomeRequest("bla bla bla"));

1	await bus.Send(new SomeRequest("bla bla bla"));

Get started

Rebus 2 is initially released as Rebus 0.90.0, along with all of its integration packages. You can install-package rebus to get started right away.

How to upgrade

Because of the many changes, upgrading from Rebus 0.84.0 (the last “Rebus 1” version) to 0.90.0 (the first “Rebus 2 beta” version) may not be trivial.

Rebus 1 endpoints do not readily work with Rebus 2 endpoints, so you will need to upgrade all endpoints at the same time or manually bridge things while you upgrade.

Bugfixes might still be released for the 0.84.0 version of Rebus, but if you have the opportunity to pick a version, I recommend you go with 0.90.0.

As always, I’ll be very grateful to hear about your experiences with it – good or bad!

Happy messaging 😀

Month: July 2015

Basic extension model in Rebus: Injectionist

Introducing Rebus 2