Producing and Consuming Messages using MassTransit and StructureMap

August 26, 2009

I am spiking out right now what it takes to use a Mass Transit as a message bus for handling a simple producer consumer messaging architecture. To be exact I want to see how easy it is to get Mass Transit working using StructureMap for an IoC container.

Dovetail uses Inversion of Control extensively to make our code more: testable and able to leap tall buildings, dodge bullets, and all that. Our favorite IoC container is StructureMap. Why? Because it is great and we have two contributors on staff: creator Jeremy Miller and the brainy Josh Flanagan.

Getting Started

My first step was to grab the Mass Transit trunk from SVN, build it and start up a new solution with three projects:

1. Producer – produces messages.
2. Consumer – consumers messages.
3. Common – code shared between producer and consumer. In this case the Message and StructureMap registry.

I included this solution view so you can see what references each project requires.

Producer

Here comes the window dressing. The Producer does one thing, generates SimpleMessages and Publishes them.

public class ProducerMain
{
    private static void Main(string[] args)
    {
        var container = new Container(x => x.AddRegistry(new CommonRegistry()));
 
        using (var bus = container.GetInstance())
        {
            do
            {
                PublishSimpleMessage(bus);
            } 
            while (Console.ReadLine().Trim() != "x");
 
            Console.WriteLine("Stopping Publisher");
        }
    }
 
    private static void PublishSimpleMessage(IServiceBus bus)
    {
        var message = new SimpleMessage {Body = Guid.NewGuid().ToString()};
 
        bus.Publish(message);
 
        Console.WriteLine("Published message : '{0}'.", message.Body);
    }
}

The interesting StructureMap setup stuff is in the CommonRegistry which is used between both Producer and Consumer projects.

public class CommonRegistry : MassTransitRegistryBase
{
    public const string MessageQueuePath = "msmq://localhost/mt-spike";
 
    public CommonRegistry() : base(typeof(MsmqEndpoint))
    {
        RegisterServiceBus(MessageQueuePath, x =>
        {
            x.ConfigureService(rs => rs.Route().To(MessageQueuePath));
        });
 
        MsmqEndpointConfigurator.Defaults(config => { config.CreateMissingQueues = true; });
    }
}

Above is the StructureMap Registry code used by both Producer and Consumer to configure MassTransit. First, we are using MSMQ for the message endpoint. Next, we are configuring a route for all SimpleMessages to be placed in the desired queue. Chris Patterson pointed me to Dru’s post on MassTransit Routing to make this work. This should not be not necessary if I was using the Subscription service. Finally, I ask Mass Transit to create any missing queues for me.

Consumer

The consumer of a SimpleMessage is a class registered with Mass Transit derived from Consumes<SimpleMessage>.All

Here is the equivalent of a Hello World message consumer:

public class SimpleMessageConsumer : Consumes.All, IBusService
{
    private IServiceBus bus;
    //private UnsubscribeAction unsubscribeAction;
 
    public void Consume(SimpleMessage message)
    {
        Console.WriteLine("Received message : '{0}'", message.Body);
    }
 
    public void Dispose()
    {
        bus.Dispose();
    }
 
    public void Start(IServiceBus bus)
    {
        this.bus = bus;
        //unsubscribeAction = bus.Subscribe(this);
        Console.WriteLine("SimpleMessage consumer started.");
    }
 
    public void Stop()
    {
        //unsubscribeAction();
    }
}

Again boring easy yet very important stuff. This consumer is where your business functionality goes. I would love to be enlightened if unsubscribe is necessary in context of how we are subscribing consumers as you’ll see next.

Automatically Registering And Subscribing Message Consumers

Before the any message consumers can receive messages they need to be registered with StructureMap and subscribed to the service bus. Take a look at the message Consumer setup:

public class ConsumerMain
{
    static void Main(string[] args)
    {
        var container = new Container(x =>
        {
            x.AddRegistry(new CommonRegistry());
            x.Scan(scan =>
            {
                scan.TheCallingAssembly();
                scan.AddAllTypesOf(typeof (Consumes<>.All));
            });
        });
 
        using (var bus = container.GetInstance())
        {
            SubscribeConsumers(container, bus);
 
            Console.ReadLine();
            Console.WriteLine("Stopping Subscriber");
        }
    }
 
    private static void SubscribeConsumers(IContainer container, IServiceBus bus)
    {
        var consumerInstances = container.Model.PluginTypes
                                .Where(p => typeof (Consumes<>.All).IsAssignableFrom(p.PluginType))
                                .SelectMany(p => p.Instances);
 
        consumerInstances.Each(type => bus.Subscribe(type.ConcreteType));
    }
}

To register all the consumers with StructureMap a scanner is used against executing assembly adding all consumers.

Next we need each consumer registered with StructureMap to be subscribed to the service bus. After this is done all consumers will now start getting any messages being published by the Producer.

The output is not too exciting. But who doesn’t like beautiful console application output?

To prove that adding more consumers was working find I created another consumer class for the same message. Nothing else was required as the configuration is automatic.

As desired. Both consumers are receiving each message being published. Also notice the beautiful Windows 7 purple window shading in the images above.

Conclusion

I hope this post enables other Mass Transit newbies like myself who are looking for a tutorial on how to get up and running. Please let me know if I could be doing anything better. I’d love to hear your feedback.

Thanks go out to Mike Hadlow as I used his A First Look at MassTransit post as a template for getting started. Even more thanks to Chris Patterson a.k.a PhatBoyG for his quick and helpful Skype support and patient answers to dumb questions.