This post details some of the internal changes to how MassTransit, an open-source lightweight service bus, communicates with transports such as MSMQ, ActiveMQ, and TIBCO. These changes are not likely to impact anyone using MassTransit, they are all well below the abstraction layer provided by the bus. At the same time, I felt it was important to share the change, along with the reasons it was made, with those that are using MassTransit.

When MassTransit was first started, MSMQ was the only transport we intended to support. In due time, however, it was determined that support for transports such as ActiveMQ and TIBCO was important. The ability to run on Linux and OS X under Mono (which does not support the System.Messaging namespace) as well as interoperability with Java systems using JMS (a specification for messaging, implemented by messaging systems like ActiveMQ and TIBCO) were the primary drivers of this decision. At the same time, insulating developers from the particulars of each transport was equally important.

To communicate with an endpoint, MassTransit uses the IEndpoint interface. The service bus would receive messages from an endpoint using this method:

IEnumerable< IMessageSelector > SelectiveReceive(TimeSpan timeout);

This involved making a call that returned an enumeration of message selectors, allowing the caller to step through the messages until an interesting message (in the case of the bus, a message with a subscribed consumer). The concerns of receiving a message were seemingly spread at random across three or four different classes (and yes, I wrote this crap). The reason for the complexity was solid though – I need the ability to selectively receive a message from a queue and skip over ones in which I have no interest.

The complexity of dealing with the yield return/break syntax of enumerators and managing scope is difficult. The programming semantics behind it are difficult to understand. I wanted something better. With all the time I’ve been spending since this was written dealing with nested closures, lambda functions, and continuations I realized there was a better way to reduce the complexity while at the same time improving extensibility.

The new signature for the receive method on an endpoint looks like this:

void Receive(Func< object, Action< object > > receiver, TimeSpan timeout);

With this new interface, the caller need only pass a method that accepts an object and returns a method that also accepts an object. The first method provides the caller an opportunity to inspect the message object to determine if the message will be consumed by the bus. If the bus is not interested, it can simply return null. If it is interested, it returns a method (either anonymous or a regular class method) that will consume the message. The endpoint will then call the returned method with the message once it has been received successfully. If the endpoint determines that the message is no longer available (if it were picked up by another process reading from the same queue for example), the returned method is not called.

The calling method looks something like this:

_endpoint.Receive(m => message => { doSomethingWith(message); });

This interface is far less complex to implement, and also made it easy to make a clean separation of what is an endpoint and what is a transport. Which leads me to…

Endpoint and Transport Split?

Sadly that reads like a Hollywood headline, but it is true. Endpoints now deal only with address resolution of sending and receiving messages and translating between the transport format and a message object (including de/serialization). New transport classes are now responsible for the actual communication with the various queue implementations supported by MassTransit.

For example, previously there was one class, MsmqEndpoint, that contained all the aspects of talking to MSMQ regardless of the type of queue (local non-transactional, local transactional, remote). Now beneath the endpoint itself, there are three MSMQ transports, one for each of these scenarios. Each of these transports cleanly deals with the particulars only, for example, the non-transactional transport has no transactional concerns in it at all.

Introducing ITransport

The new ITransport interface is narrow, dealing only with the simplest form of communication — streams. The send and receive methods from the endpoint are matched, but instead of dealing with objects, streams are used. Every transport should provide stream support at a minimum. The receive method of the transport looks like:

void Receive(Func< Stream, Action< Stream > > receiver, TimeSpan timeout);

While all transports implement streams, there is a benefit to communicating at a level above streams for certain types of endpoints. For example, when using MSMQ there are advantages to communicating directly with the Message object such as having access to the transport level message ID, the message label, and other interesting properties. To support this, the MsmqEndpoint only accepts an IMsmqTransport interface, which inherits from ITransport and adds:

void Receive(Func< Message, Action< Message > > receiver, TimeSpan timeout);

Other transports may benefit from a custom interface as well, but it is only implemented for MSMQ at this point. ActiveMQ, Loopback, and Multicast UDP all use the base stream interface.

Looking Forward

This rewrite was not purely for entertainment value (well, it was fun). Latency when sending a message from a machine to a remote queue is orders of magnitude slower than writing to a local queue. And in addition, local queues have the advantage of being local — which is important considering the first fallacy of distributed computing — the network is reliable (NOT!). To compensate for this, a more reliable method of sending messages to a remote queue is needed. By ensuring that messages sent/published by an application are durable regardless of network failure, developers can use this fire-and-forget approach to messaging that is key to building event driven applications.

To handle this, MassTransit now uses a store and forward transport for remote MSMQ queues. The store and forward transport will automatically create a local queue to cache the outbound messages destined for the remote queue. When a message is sent to the remote queue, the transport writes it to the local queue and returns to the caller. An asynchronous method then delivers the message in the background. The same transports that are used by the endpoint are reused by the store and forward transport, maintaining that high level of code reuse.

Note that on Windows Server 2003, I have observed that MSMQ will accept messages destined for a unreachable remote queue and attempt redelivery itself, but only for transactional queues (at least, that is what I have seen).

Wrapping Up

While it is always hopeful that changes like this will go by unnoticed, there is always the chance that there are some unintended consequences (read: bugs). Hopefully any of these will be weeded out quickly. In the meantime, I hope to start work on some availability features to support load balancing of command services.

In the past year, I’ve been learning about Event Driven Architecture and how using it in the enterprise can make adding functionality over time easier through a loosely-coupled event-based architecture. With Event Driven Architecture, business components can subscribe and react to events, using correlation and orchestration to combine those events to represent business processes. The loose-coupled nature makes it easier to add functionality without modifying existing services — reducing the risk of deploying new services over time.

When I first added saga support to MassTransit, the focus was purely on allowing for long-lived transactions and the orchestration of a series of events that related to a specific process. Earlier this year, the first release of a new declarative programming style for sagas using a fluent interface was put into the trunk, making it possible to design sagas using a more natural structure and language. This feature alone is one of the more compelling things about sagas in MassTransit.

Over the past month or so, I’ve been working on extending the capabilities of sagas in MassTransit beyond the orchestration of directly correlated events (represented as messages). With the latest updates to the trunk, messages that are not directly correlated to the saga can now be observed by the saga using either the interface-based or declarative state machine based saga syntax. This makes for some very compelling functionality as sagas can now observe other events and react to them!

Let’s take a quick look at the syntax using the state machine syntax using a simple message indicating a user has signed out.

public class UserSignedOut
{
	public string Username { get; set; }
}

This message only contains the username of the user that signed out of the system. If we had a saga that tracked the state of an operation on a web site based on the current user, we could observe this message and release any resources that were being tracked by the saga. For example:

public class ShoppingCartSaga : 
	SagaStateMachine< ShoppingCartSaga >
{
	static ShoppingCartSaga()
	{
		Define(()=>
		{
			Correlate(UserSignedOut).By((saga,message) => saga.Username == message.Username);
	
			During(Active,
				When(UserSignedOut)
					.Then(saga => saga.ReturnCartItemsToInventory())
					.Complete());
		};
	}

	public static Event< UserSignedOut > UserSignedOut { get; set; }
}

The Correlate() method is used to define the correlation expression used to match a message to one or more sagas in the saga repository. With this addition, the ISagaRepository has been changed to make it more integrated into the message pipeline, along with the new pipeline message sinks to support this enhanced functionality. The pipeline viewer has also been updated to show the contents of the message pipeline, including all the messages handled by the saga. The Correlate() method can also be used to change how a CorrelatedBy message is also correlated to the saga, making it possible to override the logic on how messages are used to create/use existing saga instances.

The first use case for this new functionality was to improve the subscription service. By making Subscription and SubscriptionClient sagas observe messages from each other, the sagas are able to now complete when they see a node coming back online after a crash or going offline, etc. Since the saga repositories themselves are used to keep track of all the clients and subscriptions in the system, putting that responsibility within the declaration of the saga state machine itself makes for a really clean implementation.

These bits are currently in the trunk and will be part of the next release drop of MassTransit. I encourage you to check them out and provide some feedback on any issues. Some of this may be extended soon to also allow for the custom definition of the policy associated with how messages should be applied to sagas, such as always creating a new instance or using an existing instance of a saga. Some conventions are currently in place based on the Initial state that select a default saga policy (check out ISagaPolicy and the classes that implement it for more details there).

So check it out and keep the great feedback and comments coming on the message group. The new syntax additions should be added to the wiki soon.

I’m proud to announce that we are getting close to a final 0.6 release of MassTransit. We’ve been working on this release since early December and have added a lot of new functionality. We’re also glad to have integrated a number of patches that the community has contributed — thanks for your help!

If you are currently developing with MassTransit, please check out the new bits and give us feedback. The trunk contains all of the latest code. A snapshot of the trunk, along with a 0.6 RC1 build have been added to the GoogleCode site. If you have any problems, comments, suggestions, or patches please post them to the Google Group for MassTransit. We’re interested in all experiences, good, bad, or otherwise.

A quick run down of the changes includes:

• Message Pipeline
  The message pipeline is a replacement for the message dispatcher, as well as the cache-based publishing and subscribing bits. This new pipeline is built using a pipes-and-filter approach and is dynamically adjusted as subscriptions are added and removed — both locally and by remote services.
• Saga State Machine
  With the all new method of defining sagas using a state/event fluent interface, it is now even easier to describe the business logic of a process and the events from which it is built. With the new state machine, it is not necessary to have message consumers and interfaces on your saga, you can just define the events in the class and they are wired automatically by the subscription framework. The early version of the syntax was discussed in this post.
• Better Request/Response Support
  The old bus.Request() syntax has been replaced with a new fluent style for making requests and waiting for responses. Both synchronous and asynchronous calling conventions are supported. Look for the new bus.MakeRequest extension method for more information on this slick new syntax.
• Code-based Configuration Syntax
  A new, code-based method of configuring and running a service bus has been added. This syntax significantly simplifies getting started with MassTransit and has been used throughout the unit tests. The facilities for Castle have also been updated to transparently use this new syntax with the same XML-style configuration in Windsor. More details on this syntax can be found here.
• Message Headers
  Messages are now wrapped in an envelope and headers are associated with the message. The pipeline will add certain headers, and additional headers can be added by the message publisher using an extension method to the IServiceBus.Publish and IEndpoint.Send methods. Header properties such as ResponseAddress, FaultAddress, DestinationAddress, and SourceAddress make some more advanced routing options available. All three message serializers support the message headers. Which brings me to…
• Binary, XML, and JSON Message Serializers
  All three serializers can now be configured for use as the wire message format.
• Topshelf Integration
  The old MassTransit.Host has been replaced with the easier to use Topshelf service hosting framework.
• Unsubscribe via Delegate
  The Unsubscribe methods on IServiceBus have been removed. When Subscribe is called, an UnsubscribeAction is returned instead. This delegate should be called when the subscription is no longer needed. This makes it easier to modify the state of the service bus in response to subscribe/unsubscribe actions by maintaining all of the information needed to unsubscribe in a lambda expression. The UnsubscribeAction is a delegate, and multiple unsubscribe handlers can be added together via the += operator. You can also use the extension method to get an IDisposable version of the subscription.
• Redesigned Subscription Service
  The subscription service and client have been rewritten to take advantage of the new saga syntax. This reduces the number of specialized components needed for each service, as well as leveraging the features of MassTransit internally. Check out the SubscriptionManagerGUI in the WinFormSample for a demonstration of the service, as well as the pre-built services for use in your production environment.
• Starbucks Sample Updated
  The Starbucks example has been updated to use the new saga state machines as well as some housekeeping updates to use the new features. Be sure to check it out as a starting point for understanding sagas and how to interact with them. Details on how to get the Starbucks sample running are on the wiki.

We’re hoping to get an official 0.6 release in the next couple of weeks, so please provide any feedback you have to help us wrap it up. You can get the bits, including a source and binary distribution here.