Rebus and message expiration

2012-07-14 by mookid8000 Leave a comment

If you have worked with NServiceBus, and you’ve tried Rebus as well, you might be wondering where that nifty [TimeToBeReceived] attribute went? You know, that attribute that you would decorate your messages with when you wanted to have MSMQ expire the messages for you to prevent your queues from clogging up with irrelevant messages if the recipient was down.

The answer is: There’s no such attribute in Rebus! And that’s because that would require your messages assembly to reference Rebus, and that should not be a requirement. Rebus allows your messages to be POCOs, so your messages assembly should be allowed to stay “plain” in that POCO sense.

How to set time to be received then?

With Rebus you set the time to be received property of your messages by attaching a special Rebus header to your messages: Headers.TimeToBeReceived. The value must be a HH:mm:ss string, e.g. "00:01:30" to set a 1 minute and 30 seconds expiration timeout on a message.

E.g. like so:

var somethingHappened = new SomethingHappened { Time = DateTime.UtcNow };
bus.AttachHeader(somethingHappened, Headers.TimeToBeReceived, "00:00:30");
bus.Publish(soomethingHappened);

var somethingHappened = new SomethingHappened { Time = DateTime.UtcNow };

bus.AttachHeader(somethingHappened, Headers.TimeToBeReceived, "00:00:30");

bus.Publish(soomethingHappened);

This header will be detected later on by the transport implementation (e.g. MsmqMessageQueue), which will use the approproate settings when sending the message.

In MSMQ’s case, the message will then be deleted by the queueing system when the timeout expires, which is a nifty feature that can be used to avoid flooding the queueing system with expired and irrelevant messages.

But doesn’t that require a lot of tedious manual work?

Yes – which is why I suggest you use some of the nifty hooks in Rebus to reduce the amount of tedium involved. More on that in the next post.

Contract testing with NUnit

2012-06-19 by mookid8000 4 Comments

When you’re testing your code, you’re most likely doing different types of tests – e.g. you might be doing real unit testing, integration testing, etc. In Rebus, we had the need to do contract tests[1. I’m not aware if there’s a more established word for this kind of black box test.], which is a test that ideally runs on multiple implementations of some interface, verifying that each implementation – given various preconditions in the form of differing setups etc – is cabable of satisfying a contract in the form of an interface.

In other words, for each implementation of a given interface, answer this question: Can the implementation behave in the same way in regards to the behavior we care about.

I’m aware of Grensesnitt, which Greg Young authored some time ago – and that is exactly what I’m after, only without the depending on an additional library, without the automatic stuff, etc. I just wanted to do it manually.

So, at first I wrote a lot of tedious code that would execute multiple tests for each test case, which was really clunky and tedious and would yield bad error messages when tests would fail, etc. Luckily, Asger Hallas brought my attention to the fact that a much more appropriate mechanism was already present in NUnit that would enable the thing that I wanted.

It turned out that NUnit’s [TestFixture] attribute accepts a Type as an argument, i.e. you can do this: [TestFixture(typeof(SomeType))], which will cause the test fixture type, which should be generic, to be closed with the given type. E.g. like so:

[TestFixture(typeof(SomeClassSomewhere))]
public class MyTestFixture<T>
{
    [Test]
    public void CheckType()
    {
        // so, what to do with T in here?
        Assert.That(typeof(T), Is.EqualTo(typeof(SomeClassSomewhere)));
    }
}

[TestFixture(typeof(SomeClassSomewhere))]

public class MyTestFixture<T>

{

[Test]

public void CheckType()

{

// so, what to do with T in here?

Assert.That(typeof(T), Is.EqualTo(typeof(SomeClassSomewhere)));

}

Now, if we add a generic constraint to the fixture’s type parameter, we can require that the type is a factory and can be newed up:

[TestFixture(typeof(SomeClassSomewhere))]
public class MyTestFixture<T> where T : ISomeParticularInterface, new()
{
    [Test]
    public void CheckType()
    {
        // oh wow, now we can do stuff with T
        ISomeParticularInterface t = new T();
        
        t.CallSomethingFromTheInterface();
    }
}

[TestFixture(typeof(SomeClassSomewhere))]

public class MyTestFixture<T> where T : ISomeParticularInterface, new()

{

[Test]

public void CheckType()

{

// oh wow, now we can do stuff with T

ISomeParticularInterface t = new T();

t.CallSomethingFromTheInterface();

}

See where this is going? Now one single test fixture can be closed with multiple factory types, where each factory type is capable of creating an instance of one particular implementation of something, that should be tested. E.g. in Rebus, we can test that multiple implementations of the ISendMessages and IReceiveMessages work as expected by creating a factory interface like this:

public interface ITransportFactory
{
    Tuple<ISendMessages, IReceiveMessages> Create();
    void CleanUp();
}

public interface ITransportFactory

{

Tuple<ISendMessages, IReceiveMessages> Create();

void CleanUp();

}

where an implementation might look like this:

public class RabbitMqTransportFactory : ITransportFactory
{
    RabbitMqMessageQueue sender;
    RabbitMqMessageQueue receiver;

    public Tuple<ISendMessages, IReceiveMessages> Create()
    {
        sender = new RabbitMqMessageQueue(RabbitMqFixtureBase.ConnectionString, "tests.contracts.sender", "tests.contracts.sender.error");
        receiver = new RabbitMqMessageQueue(RabbitMqFixtureBase.ConnectionString, "tests.contracts.receiver", "tests.contracts.receiver.error");
        
        return new Tuple<ISendMessages, IReceiveMessages>(sender, receiver);
    }

    public void CleanUp()
    {
        sender.Dispose();
        receiver.Dispose();
    }
}

public class RabbitMqTransportFactory : ITransportFactory

{

RabbitMqMessageQueue sender;

RabbitMqMessageQueue receiver;

public Tuple<ISendMessages, IReceiveMessages> Create()

{

sender = new RabbitMqMessageQueue(RabbitMqFixtureBase.ConnectionString, "tests.contracts.sender", "tests.contracts.sender.error");

receiver = new RabbitMqMessageQueue(RabbitMqFixtureBase.ConnectionString, "tests.contracts.receiver", "tests.contracts.receiver.error");

return new Tuple<ISendMessages, IReceiveMessages>(sender, receiver);

}

public void CleanUp()

{

sender.Dispose();

receiver.Dispose();

}

and one particular test fixture might look like this:

[TestFixture(typeof(MsmqTransportFactory))]
[TestFixture(typeof(RabbitMqTransportFactory)), Category(TestCategories.Rabbit)]
public class TestSendAndReceive<TFactory> : FixtureBase where TFactory : ITransportFactory, new()
{
    static readonly TimeSpan MaximumExpectedQueueLatency = TimeSpan.FromMilliseconds(300);

    TFactory factory;
    ISendMessages sender;
    IReceiveMessages receiver;

    protected override void DoSetUp()
    {
        factory = new TFactory();

        var transports = factory.Create();
        sender = transports.Item1;
        receiver = transports.Item2;
    }

    protected override void DoTearDown()
    {
        factory.CleanUp();
    }

    /// ....and then there's test cases down here working on sender and receiver

[TestFixture(typeof(MsmqTransportFactory))]

[TestFixture(typeof(RabbitMqTransportFactory)), Category(TestCategories.Rabbit)]

public class TestSendAndReceive<TFactory> : FixtureBase where TFactory : ITransportFactory, new()

{

static readonly TimeSpan MaximumExpectedQueueLatency = TimeSpan.FromMilliseconds(300);

TFactory factory;

ISendMessages sender;

IReceiveMessages receiver;

protected override void DoSetUp()

{

factory = new TFactory();

var transports = factory.Create();

sender = transports.Item1;

receiver = transports.Item2;

}

protected override void DoTearDown()

{

factory.CleanUp();

}

/// ....and then there's test cases down here working on sender and receiver

The advantage is that it’s much easier to see if new implementations of Rebus’ abstractions can satisfy the required contracts – and with the way it’s implemented now, various implementations can have different setup and teardown code, and their fixtures can belong to a suitable NUnit category, allowing Rebus’ build servers to run only tests where the required infrastructure is in place.

Official Rebus documentation

2012-06-11 by mookid8000 Leave a comment

Rebus logo Hello world!

Due to the fact that there’s actually a few people using Rebus in real systems now, and also to give myself some insight into whether I have thought things through ;), I have started working on the official Rebus documentation.

For now, it resides on Rebus’ GitHub wiki – if you’re interested, you should check it out.

Also, if you think that something is missing, please don’t hesitate to let me know!

Slides from my Community Day 2012 talk about Rebus

2012-05-10 by mookid8000 Leave a comment

Today, I did a talk on Rebus. It’s in Danish, and the slides can be seen below 🙂

Rebus er ingen gåde

View more presentations from Mogens Heller Grabe

Check out The Snoop

2012-03-21 by mookid8000 2 Comments

It’s very very alpha, even more alpha than Rebus itself – but here it is: Rebus Snoop – a tool to aid developers and operations in managing their Rebus queues.

The goal of The Snoop is to provide that little extra bit of tooling that will give you a feeling of transparency and confidence when developing and operating systems based on Rebus.

At the moment, it’s extremely basic – it allows you to see which queues exist on a given machine, and it will display the contents of the selected queue.

When a message is selected, various pieces of information from the message will be shown in the panel to the right.

If the extension element from the MSMQ message can be properly deserialized into a dictionary of headers, and those headers are proper Rebus headers, they will be displayed at the top.

If the rebus-encoding header is present, The Snoop will attempt to decode the message body and show the contents in the windows to the right.

If the rebus-source-queue header is present, The Snoop will unlock the “Return to source queue” button, allowing you to retry the selected messages.

You can go to Rebus’ wiki page on the Snoop if you want to read more about The Snoop.

I’d be grateful if you would report any problems you may encounter and/or crazy feature requests to Rebus’ issue tracker.

How to get up and running with Rebus

2012-02-26 by mookid8000 2 Comments

Now that Rebus is present on NuGet, it has become extremely easy to get started. This small guide will show how to build a simple TopShelf service, that will send a message to itself. It should be pretty easy to extend this to a more complete example with multiple services, but let’s just keep it really simple for now.

Create the host process

First, we create a TopShelf host process that has all the necessary stuff to be able to run and be debugged, as well as installed, uninstalled, and be managed as a Windows Service. Precious stuff when you’re building backend processes.

Create a Console Application

Do what the header says – and change the target framework from that bastard “.NET 4 Framework Client Profile” to “.NET 4 Framework”.

Go get TopShelf

Open a package manager console (we’ll need it for the Rebus alpha packages), and

Install-Package TopShelf -Project <NameOfYourProject>
Install-Package Log4Net -Project <NameOfYourProject>

1 2	Install-Package TopShelf -Project <NameOfYourProject> Install-Package Log4Net -Project <NameOfYourProject>

For some reason, you need to download Log4Net manually, even though TopShelf depends on it.

Configure Log4Net and declare the TopShelf service

Now, do something like this in your Program.cs:

using Topshelf;
using log4net.Config;

class Program
{
    static void Main()
    {
        XmlConfigurator.Configure();

        HostFactory
            .Run(c =>
                        {
                            c.SetServiceName("RebusTest");
                            c.SetDisplayName("RebusTest");
                            c.Service<Program>(p =>
                                                {
                                                    p.ConstructUsing(() => new Program());
                                                    p.WhenStarted(s => s.Start());
                                                    p.WhenStopped(s => s.Stop());
                                                });
                        });
    }

    void Start()
    {
    }

    void Stop()
    {
    }
}

using Topshelf;

using log4net.Config;

class Program

{

static void Main()

{

XmlConfigurator.Configure();

HostFactory

.Run(c =>

{

c.SetServiceName("RebusTest");

c.SetDisplayName("RebusTest");

c.Service<Program>(p =>

{

p.ConstructUsing(() => new Program());

p.WhenStarted(s => s.Start());

p.WhenStopped(s => s.Stop());

});

}

void Start()

{

}

void Stop()

{

}

and then add some XML to your app.config:

  <configSections>
    <section name="log4net" type="log4net.Config.Log4NetConfigurationSectionHandler, log4net" />
  </configSections>

  <log4net>
    <appender name="ColoredConsoleAppender" type="log4net.Appender.ColoredConsoleAppender">
      <mapping>
        <level value="ERROR" />
        <foreColor value="White" />
        <backColor value="Red" />
      </mapping>
      <layout type="log4net.Layout.PatternLayout">
        <conversionPattern value="%date [%thread] %-5level %logger [%property{NDC}] - %message%newline" />
      </layout>
    </appender>
    <root>
      <priority value="DEBUG" />
      <appender-ref ref="ColoredConsoleAppender" />
    </root>
  </log4net>

</configSections>

</mapping>

</layout>

</appender>

<root>

<appender-ref ref="ColoredConsoleAppender" />

</root>

</log4net>

Now you should be able to run the service and see a buttload of log statements, courtesy of TopShelf.

Insert Rebus

Now that we have the host process up and running, let’s make the program do stuff.

Let’s get Rebus

Go to the package manager console again, and do this:

Install-Package Rebus.Log4Net -Pre -Project <NameOfYourProject>
Install-Package Rebus.Castle.Windsor -Pre -Project <NameOfYourProject>

1 2	Install-Package Rebus.Log4Net -Pre -Project <NameOfYourProject> Install-Package Rebus.Castle.Windsor -Pre -Project <NameOfYourProject>

Let’s configure and use it!

Now, you should be able to implement your Start and Stop methods like this:

IBus bus;
IWindsorContainer container;

void Start()
{
    container = new WindsorContainer();

    container.Register(Component.For<IHandleMessages<string>>().Instance(this));

    bus = Configure.With(new WindsorContainerAdapter(container))
        .Logging(l => l.Log4Net())
        .Transport(t => t.UseMsmqAndGetInputQueueNameFromAppConfig())
        .Serialization(s => s.UseJsonSerializer())
        .DetermineEndpoints(d => d.FromRebusConfigurationSection())
        .CreateBus()
        .Start();

    bus.Send("HELLLLLO WORLD!!!1");
}

void Stop()
{
    container.Dispose();
}

IBus bus;

IWindsorContainer container;

void Start()

{

container = new WindsorContainer();

container.Register(Component.For<IHandleMessages<string>>().Instance(this));

bus = Configure.With(new WindsorContainerAdapter(container))

.Logging(l => l.Log4Net())

.Transport(t => t.UseMsmqAndGetInputQueueNameFromAppConfig())

.Serialization(s => s.UseJsonSerializer())

.DetermineEndpoints(d => d.FromRebusConfigurationSection())

.CreateBus()

.Start();

bus.Send("HELLLLLO WORLD!!!1");

}

void Stop()

{

container.Dispose();

}

Now, in order to make the container.Register(...) line from above compile, you need to let Program implement IHandleMessages<string>. Try adding something like this:

class Program : IHandleMessages<string>
{
    // (... cut for brevity ....)

    public void Handle(string message)
    {
        Console.WriteLine("w00t!!!: {0}", message);
    }
}

class Program : IHandleMessages<string>

{

// (... cut for brevity ....)

public void Handle(string message)

{

Console.WriteLine("w00t!!!: {0}", message);

}

If you run the program now, you’ll get some kind of error, saying that the input queue name is invalid. That’s because we haven’t configured Rebus yet. Let’s do that by modifying app.config by adding this:

<?xml version="1.0" encoding="utf-8"?>
<configuration>
  <configSections>
    <!-- (...) -->

    <section name="rebus" type="Rebus.Configuration.RebusConfigurationSection, Rebus" />
  </configSections>

  <!-- (...) -->

  <rebus inputQueue="RebusTest.input" workers="2">
    <endpoints>
      <add messages="System.String, mscorlib" endpoint="RebusTest.input" />
    </endpoints>
  </rebus>

</configuration>

<?xml version="1.0" encoding="utf-8"?>

</configSections>

</endpoints>

</rebus>

</configuration>

The InputQueue attribute above tells Rebus to receive messages from the RebusTest.input queue, and the <add ...> elements says to send messages of the System.String type to the same RebusTest.input queue.

Let’s run it

Now you should be able to run it and see something like this on your screen:

If you get the same w00t!!!: HELLLLLO WORLD!!!1 in your console, it means you’re ready to start building stuff with Rebus 🙂

Announcing: Rebus

2012-02-26 by mookid8000 3 Comments

Preface

Friends and acquaintances might know that I’ve spent the last 6 months or so of my pastime, hacking on something I call Rebus. To cut a long story short, I can tell you that I am a happy NServiceBus user who became very sad when NServiceBus went from being free to requiring a commercial license – not because I think NServiceBus isn’t worth spending money on, but because I think it hurts the adoption of NServiceBus.

Speaking for myself, I have already refrained from using NServiceBus in a couple of smaller projects, where it would otherwise have been an awesome addition.

I seriously considered forking the 2.0 version of NServiceBus, which is licensed under Apache v. 2.0, but I was overwhelmed by the sheer size of the project. This led me to re-implement my favorite features of NServiceBus as Rebus, trying to focus and stay lean along the way.

And here it is: Rebus

Rebus – the core – depends on .NET 4 only. And since .NET 4 has MSMQ, ADO.NET, and binary serialization, you get to send and receive messages transactionally and persist your stuff in SQL Server without anything but Rebus core.

3rd party integration (like e.g. MongoDB, NewtonSoft JSON serializer, etc) is provided through small, dedicated add-on projects. This granularity makes it easier to manage your own dependencies.

Update: And since JSON serialization was deemed to be the preferred form of message serialization, it has been ILMerged into Rebus core since right after this announcement. Therefore, JSON serialization is included in the box and is the default.

There’s currently two ways to get started with Rebus:

Instantiate RebusBus manually, filling in the ginormous constructor with implementations for how you want the bus to run.
Use the configuration API.

Let’s try the configuration API – with the following snippet of XML in your app.config:

<?xml version="1.0" encoding="utf-8" ?>
<configuration>
  <configSections>
    <section name="log4net" type="log4net.Config.Log4NetConfigurationSectionHandler, log4net"/>
    <section name="rebus" type="Rebus.Configuration.RebusConfigurationSection, Rebus" />
  </configSections>

  <rebus inputQueue="myService.inputQueue" workers="5">
    <endpoints>
      <add messages="AnotherService.Messages" endpoint="anotherService.inputQueue"/>
    </endpoints>
  </Rebus>

  <log4net>...</log4net>
</configuration>

<?xml version="1.0" encoding="utf-8" ?>

</configSections>

</endpoints>

</Rebus>

</configuration>

and the following piece of code in your app:

var myAdapter = new MyFavoriteContainerAdapter(myFavoriteContainer);

Configure.With(myAdapter)
    .Logging(c => c.Log4Net())
    .DetermineEndpoints(c => c.FromRebusConfigurationSection())
    .Transport(c => c.UseMsmqAndGetInputQueueNameFromAppConfig())
    .Serialization(c => c.UseJsonSerializer())
    .CreateBus()
    .Start();

// IBus is in the container now :)

var myAdapter = new MyFavoriteContainerAdapter(myFavoriteContainer);

Configure.With(myAdapter)

.Logging(c => c.Log4Net())

.DetermineEndpoints(c => c.FromRebusConfigurationSection())

.Transport(c => c.UseMsmqAndGetInputQueueNameFromAppConfig())

.Serialization(c => c.UseJsonSerializer())

.CreateBus()

.Start();

// IBus is in the container now :)

you should be up and running with a Rebus bus.

That was a small teaser – there’s more blog posts on the way, I promise 😉

So, what’s the state of this?

Well, that’s actually the whole point of this blog post – I’m currently in a situation where I really feel like putting Rebus to some serious use, but I cannot justify foisting it on a client until I have a big project to prove its worth. And I cannot prove its worth in a big project until I have a client I can fob it on.

That’s where you come in! 🙂

You:

You’re building something – it’s not totally mission critical, as in human lives depend on it, but it’s still something.
You want some nifty service bus technology to help you build a loosely coupled architecture that will allow your project to flex and bend and integrate with stuff in many years to come.
You wanted to use NServiceBus because it’s cool, but you can’t afford the cost up front.
You’re reading this blog post, or someone you know read it and is now telling you about it…

If that’s you, then you should meet Rebus!

Rebus:

Absolutely free to use and abuse.
Very very similar to NServiceBus, allowing you to migrate to NServiceBus some time in the future if Rebus is not enough for you.
Pretty simple.
Comes with unlimited[1. Within reason, of course – I have a day job, a wife, and two kids, so I’m not answering emails 24/7 – but I’ll go to great lengths to help you get a smooth adoption of Rebus – because that’s just how much I care :)] support from its author.
Did I mention it was free?

A match made in heaven?

If you have the slightest bit of interest in what I’m suggesting, please contact me – either via email or on Twitter – and then we’ll talk about what Rebus can do for you. Oh, and please don’t hesitate to contact me – even if you’re NOT planning on using Rebus, I’d like to know why you picked NServiceBus, MassTransit, or Rhino Service Bus instead.

If you’re just eager to try it out, feel free to Install-Package -Pre one or more of Rebus’ NuGet packages.

The duality of request/reply vs. publish/subscribe #2

2012-01-23 by mookid8000 Leave a comment

In my last post, I described how the mechanics of publish/subscribe actually mirror those of request/reply.

In this post, I’ll look at the two operation from another angle: What do they mean?

What does it mean when you bus.Send?

Sending means either that the sender wants to

Command that another service does something.
Request that another service does something, and yields one or more replies[1. Note that the request/reply pattern may impose unwanted temporal coupling in an architecture and should probably be used only in integration scenarios orchestrated by a saga.].

This means that the sender knows stuff about the other service, but that other service will most likely not know or care about who’s sending. In other words, the sender depends on that other service!

What does it mean when you bus.Publish?

Publishing means that the publisher wants to

Broadcast an event that contains information on something that has happened.

This means that the publisher most likely does stuff inside itself, maybe updates some internal state, and then goes on and publishes information on some aspect of what has happened. In doing this, the publisher will most likely not know or care about who’s receiving. In other words, the subscriber depends on the publisher!

Summing it up with a picture

Consider this illustration, where service dependencies are shown with arrows:

Again, see how comparing Send to Publish is actually like comparing two mirror images when the other mirror image is upside-down?

The duality of request/reply vs. publish/subscribe #1

2012-01-20 by mookid8000 Leave a comment

A question I often meet in relation to messaging frameworks like NServiceBus and Rebus, is this: Where do messages go?

The confusion often comes from comparing how bus.Publish works with how bus.Send works.

In this post, I’d like to describe the two operations and show that they are mirror images of each other – except maybe not as much a mirror image as a transposition.

Sending messages

In the case where you’re doing a bus.Send(message), the answer is trivial: The message gets sent to the endpoint specified in the sender’s enpoint mapping for that message type. Let’s say our sender is equipped with this snippet of XML[1. The snippet is an endpoint mapping in NServiceBus format, which can also be understood by Rebus when it’s running with the DetermineDestinationFromNServiceBusEndpointMappings implementation of IDetermineDestination] in its app.config:

<UnicastBusConfig>
  <MessageEndpointMappings>
    <add Messages="MyService.Messages" Endpoint="my_service"/>
  </MessageEndpointMappings>
</UnicastBusConfig>

</MessageEndpointMappings>

</UnicastBusConfig>

If we assume that message is an instance of a class from the MyService.Messages assembly, in this case a bus.Send(message) will be translated into bus.Send("my_service", message).

Publishing messages

But where do messages go when they’re published? Well, they go to whomever subscribed to that particular message type – and with NServiceBus (and, for that matter, with Rebus as well) subscribers get subscribed by sending some kind of subscription message, which is basically saying: “Hey there, mr. Publisher – I’m some_subscriber, and I’d like to subscribe to MyService.Messages.SomeParticularMessage”.

From this point on, the publisher will store the mapping of the message type along with the subscriber’s address, allowing a bus.Publish(message) method to be implemented something along the lines of

public void Publish(object message)
{
    foreach(var subscriberEndpoint in GetSubscribersFor(message.GetType()))
    {
        Send(subscriberEndpoint, message);
    }
}

public void Publish(object message)

{

foreach(var subscriberEndpoint in GetSubscribersFor(message.GetType()))

{

Send(subscriberEndpoint, message);

}

So – how do we understand this on a higher level, allowing us to never ever confuse these things again? Let’s dive into…

The duality

Consider these two sequence-like diagrams:

See how request/reply and publish/subscribe are actually the same patterns? The reason these two are often confused, is that the Send operation is often countered by Publish, when in fact it would be more fitting to see the subscription message (i.e. subscription request) as the counterpart of Send. Thus, Publishing is more like replying. And thus, Send is actually the transposition of Publish.

Now, when you realize this, you’re never going to confuse these things again 🙂 In the next post, I’ll touch a little bit on another difference between Send and Publish.