I just finished doing a second de-optimization for NMemcached. This is the second such change I do, turning the code from using the async read pattern to using a simple serial read from a stream. Both those changes together takes the time to complete my simple (and trivial) benchmark from 5768.5 ms to 2768.6 ms.

That is less than 50% of the original time! In both cases, I started with high use of BeginXyz, in order to get as much parallelism as much as possible, but it actually turned out to be a bad decision, since it meant that in many cases where the data was already there, I would pay the price of an async call, vs. just grabbing the data from the kernel buffer.

Udi is talking about occasionally connected clients, and point to several challenges that you have to face when you are building such applications. One of the major issues with such clients is that you have to take into account the fallacies of distributed computing and how to deal with them.

But, and this is important, having to deal with those can be expensive in terms of time and complexity. There are many scenarios when telling the user that the system is not operational is a valid choice. No, it is not always valid, and there are as many application where you explicitly need to deal with that, but there are more when connectivity can be assumed and the only worry you have in this regard is handling the failure condition gracefully.

Last night I found myself up at 1 AM, feeling restless. I decided that I need a new project, something that is very simple and will allow me to channel some energy. I decided to pursue something that I have long been curious about, build a Memcached server in .Net. The Memcached server is a distributed cache server that is very popular in the Ruby, PHP & Perl worlds, with some following on both Java & .Net side.

It is written in C, and it is has some fairly interesting characteristics. Chief among them, it is a straightforward technical challenge. I spend about a day on this, but I have a working Memcached server, which can work with the existing clients.

Currently I support all the standard commands:

• add
• append
• cas
• delete
• prepend
• replace
• set
• get
• gets
• incr
• decr
• flush_all
• quit
• version

The only thing that I do not support is the stats command. The project has over a hundred unit tests and 34 integration tests and currently stand at 94% test coverage.

All of that said, what will probably interest most people is the performance comparison. I do hope that some people will stop to actually look at the design of the code, but here are the numbers, for reading / writing of 10,000 (small) values (over 10 iterations, using 20 connections):

Native Memcached finish in: 1709.6 ms

NMemcached completes in: 5768.5 ms

Update:

Based on profiling, I decided to make a tiny change to the part of the application that handles reading the commands from the user. Now it is reading them in continuous fashion, instead of one byte at a time. This brought the speed of the NMemcached version to 3144.5 ms, which is a huge benefit for such a small change.

Which put a project that was written in a day at just over 3 times slower than a heavily optimized piece of mature C code. Not bad, if I say so myself. I took a look with dot Trace, to see what was taking most of the time, and it looks like a significant amount goes in the networking section (specifically the LineReader class). There are a lot of tests, so it is not a problem to go and fix this if someone wants to.

You can access the code here: https://rhino-tools.svn.sourceforge.net/svnroot/rhino-tools/experiments/NMemcached

This piece of code make me so happy:

private static IDictionary<string, Func<Stream, ICommand>> commandFactories =
	new Dictionary<string, Func<Stream, ICommand>>
		{
			//storage
			{"add", s => new AddCommand(s)},
			{"append", s => new AppendCommand(s)},
			{"cas", s => new CasCommand(s)},
			{"delete", s => new DeleteCommand(s)},
			{"prepend", s => new PrependCommand(s)},
			{"replace", s => new ReplaceCommand(s)},
			{"set", s => new SetCommand(s)},

			// retrieval
			{"get", s => new GetCommand(s)},
			{"gets", s => new GetsCommand(s)},

			//modifications
			{"incr", s => new IncrCommand(s)},
			{"decr", s => new DecrCommand(s)},

			//misc
			{"flush_all", s => new FlushAllCommand(s)},
			{"quit", s => new QuitCommand(s)},
			{"version", s => new VersionCommand(s)},
		};

This is something that I do quite often, and it is good to have a way to do it easily.

Here are a few interesting tests that I just wrote:

[TestFixture]
public class SystemWebCacheTests : CacheMixin
{
	[SetUp]
	public void Setup()
	{
		ClearCache();
	}

	[Test]
	public void IterationWithModifications()
	{
		Cache["foo"] = new object();
		Cache["bar"] = new object();

		foreach (DictionaryEntry de in Cache)
		{
			Cache.Remove(de.Key.ToString());
		}
		Assert.AreEqual(0, Cache.Count);
	}

	[Test]
	public void IterationWithAdditions_ShouldGet_ConsistentSnapshot()
	{
		Cache["foo"] = new object();
		Cache["bar"] = new object();
		bool first = true;
		int count = 0;
		foreach (DictionaryEntry de in Cache)
		{
			count += 1;
			if(first==false)
				Cache["baz"] = new object();
			first = false;
		}
		Assert.AreEqual(2, count);
		Assert.AreEqual(3, Cache.Count);
	}

	[Test]
	public void IterationWithRemoval_ShouldGet_ConsistentSnapshot()
	{
		Cache["foo"] = new object();
		Cache["bar"] = new object();
		bool first = true;
		int count = 0;
		foreach (DictionaryEntry de in Cache)
		{
			count += 1;
			if (first)
				Cache.Remove("bar");
			Assert.IsNotNull(de.Value);
			Assert.IsNotNull(de.Key);
			first = false;
		}
		Assert.AreEqual(2, count);
		Assert.AreEqual(1, Cache.Count);
	}
}

I have no idea if this is just a coincidental evidence, but I like the consistency.

Just got a rather pointed reminder why you should try to get as many tests as possible. The code is:

public override void Execute()
{
	Version version = typeof (VersionCommand).Assembly.GetName().Version;
	Writer.WriteLine("VERSION " + version);
}

And the test is:

[Test]
public void Will_return_assembly_version_as_memcached_version()
{
	var stream = new MemoryStream();
	var cmd = new VersionCommand(stream);
	cmd.Execute();
	Assert.AreEqual("VERSION " + typeof(VersionCommand).Assembly.GetName().Version 
		+ "\r\n", ReadAll(stream));
}

Now, tell me why this is not a stupid test & useless test.

For a few days, some (~4) of SvnBridge integration tests would fail. Not always the same ones (but usually the same group), and only if I run them all as a group, never if I run each test individually, or if I run the entire test class (which rules out most of the test dependencies that causes this). This was incredibly annoying, but several attempts to track down this issue has been less than successful.

Today I got annoyed enough to say that I am not leaving until I solve this. Considering that a full test run of all SvnBridge's tests is... lengthy, that took a while, but I finally tracked down what was going on. The fault was with this method:

protected string Svn(string command)
{
	StringBuilder output = new StringBuilder();
	string err = null;
	ExecuteInternal(command, delegate(Process svn)
	{
		ThreadPool.QueueUserWorkItem(delegate

		{
			err = svn.StandardError.ReadToEnd();
		});
		ThreadPool.QueueUserWorkItem(delegate
		{
			string line;
			while ((line = svn.StandardOutput.ReadLine()) != null)
			{
				Console.WriteLine(line);
				output.AppendLine(line);
			}
		});
	});
	if (string.IsNullOrEmpty(err) == false)
	{
		throw new InvalidOperationException("Failed to execute command: " + err);
	}
	return output.ToString();
}

This will execute svn.exe and gather its input. Only sometimes it would not do so.

I fixed it by changing the implementation to:

protected static string Svn(string command)
{
	var output = new StringBuilder();
	var err = new StringBuilder();
	var readFromStdError = new Thread(prc =>
	{
		string line;
		while ((line = ((Process)prc).StandardError.ReadLine()) != null)
		{
			Console.WriteLine(line);
			err.AppendLine(line);
		}
	});
	var readFromStdOut = new Thread(prc =>
	{
		string line;
		while ((line = ((Process) prc).StandardOutput.ReadLine()) != null)
		{
			Console.WriteLine(line);
			output.AppendLine(line);
		}
	});
	ExecuteInternal(command, svn =>
	{
		readFromStdError.Start(svn);
		readFromStdOut.Start(svn);
	});

	readFromStdError.Join();
	readFromStdOut.Join();

	if (err.Length!=0)
	{
		throw new InvalidOperationException("Failed to execute command: " + err);
	}
	return output.ToString();
}

And that fixed the problem. What was the problem?

The Umbrella project from nVentive. I had the chance of sitting with Francois Tanguay in DevTeach and he showed me a bit about it. I was impressed. Umbrella is the ultimate utility library, using all the advantages C# 3.0 can give it. In fact, I think of it a bit like boost for C#.

I tried to read the code, and it was hard. Because of the util nature of Umbrella, there isn't a lot of context around the code, so you have to parse the code to understand what it is doing. Luckily, there are a lot of unit tests, which allows me to get the context. I strongly recommend reading the tests in stead of the code, they are a far better way to get an understanding on what the project can offer.

I'll go with my usual style, top to bottom, and review what I can.

Binding

This is a concept that I am familiar with from boost. It is also called memoization currying, and is used frequently in functional languages. Here are a few tests, which would explain it much better:

[Fact]
public void Bind_With_One_Param()
{
	Func<int, int> echo = (i) => i;

	var always10 = echo.Bind(10);
	Assert.Equal(10, always10());
}

[Fact]
public void Bind_First_With_Two_Params()
{
	Func<double, double, double> power = Math.Pow;

	var twoToTheN = power.BindFirst(2.0);
	Assert.Equal(1024.0, twoToTheN(10.0));
}

[Fact]
public void Bind_Chaining()
{
	Func<int, int, int, int, long> func = (a, b, c, d) => a + b + c + d;

	var chain = func.BindFirst(1).BindFirst(2).BindFirst(3).Bind(4);
	Assert.Equal(func(1, 2, 3, 4), chain());
}

If you are working with delegates a lot, this is a really nice way to handle that.

Clock

This is a simple abstraction (IClock, SystemClock, FreezedClock) over the current time, which allows you to play around with time with total disregard to the system time. Basically, an abstraction on top of DateTime.Now. The main goal is usually to enable unit testing, but I have found this approach to be very useful for handling out of time processing. What does this mean? If I want to process payroll for February, I should be able to do so, and all calculations should be set to whatever date I am talking about, not the current date. (In fact, in such systems, there is usually the target date and effective date, both of which are used to calculate various things, but I digress.)

Collections

Collections are important, no doubt about that, and Umbrella has a lot to offer there. The first interesting example is automatic conversion between types, supported by the CollectionAdapter. Here is the test that shows how this works:

[Fact]
public void CanConvertBetweenCollectionsTypes()
{
	var ints = new List<int>();
	ICollection<string> strings = new CollectionAdapter<int, string>(
		ints,
		Funcs<string, int>.Convert,
		Funcs<int, string>.Convert);

	strings.Add("1");
	Assert.Equal(1, ints[0]);
}

Note the Funcs.Convert calls, we will discuss them later. The abstraction is quite complete and very useful. This is also a good way of implementing covariance with generics, I think.

There is a wealth of extension methods for dealing with collections. The most important one, as far as I am concerned, is the AddRange over ICollection<T>. I keep wanting that, and it is only there for List<T>. Having it as an extension method is sweet.

[Fact]
public void AddRange_And_ReplaceWith()
{
    ICollection<int> collection = new List<int>();
    

    collection.AddRange(new int[] { 1, 2 });

    collection.ReplaceWith(new int[] { 3, 4 });

    Assert.Equal(2, collection.Count);
    Assert.Equal(3, collection.ElementAt(0));
    Assert.Equal(4, collection.ElementAt(1));
}

Another interesting extension method is Subscribe. I don't like the name, because it has nothing to do with what the code does, but it is a really nice idea nonetheless:

[Fact]
public void Subscribe()
{
	using (collection.Subscribe(1))
	{
		Assert.Equal(1, collection.Count);
	}
	Assert.Empty(collection);
}

Inside the using block, the collection has the element, but during dispose, it is removed. I think that the reason it is called this way is that it is being used by ObservableExtensions (later).

Dictionary also get some love, with FindOrCreate and GetValueOrDefault, both of which are very welcome. Enumerable get a much needed ForEach extension method, as well as other interesting bits, ranging from bool None(predicate) to checking if it is empty to IndexOf, etc. LazyList make an appearance, as it should.

What I find very interesting is SyncronizedDictionary. It is interesting because of the way it is implemented, and because dictionaries are almost always a source of hard to realize threading issues in many applications. The implementation made me laugh, however:

public TValue this[TKey key]
{
    get { return Lock.Read(item => item[key]); }
    set { Lock.Write(item => item[key] = value); }
}

It made me laugh because it is so simple. I traced down the implementation, and I liked it. It comes down to using Reader Writer Lock, which is what you are supposed to be doing. Obvious Umbrella has attained its critical mass.

Components

Ubmrella is also big enough to include its own Service Locator implementation, which can be plugged into an IoC container. I think that I'll skip this part.

Composites

All the building blocks for the composite patterns are here, and in a very interesting fashion. Take a look at this:

public class CompositeFoo : Composite<IFoo>, IFoo
{
    #region IFoo Members

    public void Do(int i)
    {
        Items.ForEach(item => item.Do(i));
    }

    #endregion
}

And now we can use this like this:

[Fact]
public void SelectMany()
{
	var foo = new CompositeFoo
	          	{
	          		new Foo(),
	          		new OtherFoo(),
	          		new OtherFoo(),
	          		new CompositeFoo
	          			{
	          				new Foo(),
	          				new OtherFoo()
	          			}
	          	};
	IEnumerable<IFoo> items = foo.SelectMany();

    Assert.Equal(5, items.Count());
    Assert.Equal(2, items.OfType<Foo>().Count());
    Assert.Equal(3, items.OfType<OtherFoo>().Count());
}

I don't use composites all that much, but it is an elegant approach.

Conditions

You can almost classify this as an extension of the composite pattern, by allowing to compose conditions. This is really useful when you have to build complex conditions are runtime.

[Fact]
public void True()
{
    IMessage<Null, bool> lhs = new Message<Null, bool>(notUsed => true);
    IMessage<Null, bool> rhs = new Message<Null, bool>(notUsed => true);

    var andMessage = lhs.And(rhs);

    Assert.True(andMessage.Send());
}

However, did you note the IMessage? I will get to that in a bit, and we will discuss this, it is an important part of Umbrella, but not one I like.

Containers

Couldn't figure out what this is supposed to do. Seems to be relation to the service locator impl.

Contracts

Couldn't figure out what this is supposed to do. Seems to be relation to the service locator impl.

Conversions

There are some nice things there, mostly accessible via Conversion().To<TTarget>(), handling Enums is nice:

[Fact]
public void StringFromEnum()
{
    Assert.Equal("Cancelled", Status.Cancelled.Conversion().To<string>());
    Assert.Equal("VER", Status.Verified.Conversion().To<string>());
}

public enum Status
{
    [Description("VER")]
    Verified,
    Cancelled
}

Or standard type conversions:

[Fact]
public void CanConvertInt32ToString()
{
    Assert.Equal("1", 1.Conversion().To<string>());
}

[Fact]
public void CanConvertStringToInt32()
{
    Assert.Equal(1, "1".Conversion().To<int>());
}

Decorator

Again, the building blocks for the decorator pattern. This time, I don't see much use of this, since it doesn't seem to provide much value.

Equality

Provide a set of extension methods that extend equality comparisons. Mostly over enumerables and the like, from a brief look.

Events

There are the expected Raise() set of extension methods, but I find Observe and Notify far more interesting:

using(order.Observe(UpdateForm))
{
	form.Show();
}

And:

public int Value
{
    get { return value; }
    set
    {
        this.value = value;
        this.Notify(PropertyChanged, item => item.Value);
    }
} 

During the using statement, all NotifyPropertyChanged events will be captured. This is nice. The Notify() will handle property notifications without using strings. I don't deal much with INotifyPropertyChanged, however, so I care little for this.

There is also an implementation of observable, which may be useful. Again, this is not something I tend to use.

Expressions

Linq expressions are immutable, which make it a harder to build them. Umbrella solve this by giving us editable expressions, which are builders on top of normal extensions. Nice.

Extensions

This is a very loaded term in a utility project, and it contains a lot of things.  Some of the things I liked are date handling:

[Fact]
public void Equal()
{
    DateTime x = new DateTime(2008, 2, 1);
    DateTime y = new DateTime(2008, 2, 2);

    Assert.True(x.Equal(y, DateTimeUnit.ToMonth));
    Assert.False(x.Equal(y, DateTimeUnit.ToDay));
}

[Fact]
public void Truncate()
{
    Assert.Equal(new DateTime(2008, 2, 1), new DateTime(2008, 2, 2).Truncate(DateTimeUnit.ToMonth));
}

There are a lot of things around date, but I think that this is a representative method:

public static DateTime BeginningOfWeek(this DateTime self)
{
	return (self - self.DayOfWeek.DaysSince(Extensions.WeekBeginsOn).Days())
				.BeginningOfDay();
}

There are also the standard (by now), "foo".IsNullOrEmpty() and "foo".HasValue(), additions to

I can't figure out the concept of Extension Points, however. Or, to be rather exact, I can't see what value it brings to the table.

What is really sweet is the handling of bit manipulation:

[Fact]
public void Enum_Add()
{
    BindingFlags flags = BindingFlags.Public;
    flags = flags.Add(BindingFlags.NonPublic);

    Assert.Equal(BindingFlags.Public | BindingFlags.NonPublic, flags);
}

[Fact]
public void Enum_Remove()
{
    BindingFlags flags = BindingFlags.Public | BindingFlags.NonPublic;
    flags = flags.Substract(BindingFlags.NonPublic);

    Assert.Equal(BindingFlags.Public, flags);
}

And the correspond read approach:

[Fact]
public void Enum_ContainsAll()
{
    BindingFlags flags = BindingFlags.Public | BindingFlags.NonPublic;

    Assert.True(flags.ContainsAll(BindingFlags.Public));
    Assert.True(flags.ContainsAll(BindingFlags.NonPublic));
    Assert.True(flags.ContainsAll(BindingFlags.Public | BindingFlags.NonPublic));

    Assert.False(flags.ContainsAll(BindingFlags.Public | BindingFlags.Instance));
    Assert.False(flags.ContainsAll(BindingFlags.Instance));

}

[Fact]
public void Enum_ContainsAny()
{
	BindingFlags flags = BindingFlags.Public | BindingFlags.NonPublic;

	Assert.True(flags.ContainsAny(BindingFlags.Public));
	Assert.True(flags.ContainsAny(BindingFlags.NonPublic));
	Assert.True(flags.ContainsAny(BindingFlags.Public | BindingFlags.NonPublic));
	Assert.True(flags.ContainsAny(BindingFlags.Public | BindingFlags.Instance));

	Assert.False(flags.ContainsAny(BindingFlags.Instance));
}

Factories

Seems to be relation to the service locator impl.

Locator

This is the service locator impl, I am ignoring that.

Messages

Messages are pretty important in Umbrella. They are defined as:

public interface IMessage<TRequest, TResponse>
{
    TResponse Send(TRequest request);
}

I strongly disagree with the term message here, however. A message in inanimate object, it doesn't act on its own. Messages in Umbrella are actions. In fact, the default implementation of a message is around a Func<TRequest, TResponse>.

Once you have those messages, however, you can start dealing with them in interesting ways. As a simple example, you can create a DisposableMessage, which will wrap another message and call Send() on it. Or bind the output of one message to the input of another, or simply chain them all together. A lot of the actions inside Umbrella are happening using those messages.

Reflection

This is a probably my favorite part in Umbrella. It abstract away all the gory reflection details:

[Fact]
public void Instance()
{
    Foo foo = new Foo();

    IReflectionExtensionPoint fooReflection = foo.Reflection();

    Assert.Equal(foo.I, fooReflection.Get("i"));

    fooReflection.Set("i", 2);

    Assert.Equal(2, foo.I);

    Assert.Equal(2, fooReflection.Get("I"));

    fooReflection.Set("I", 3);

    Assert.Equal(3, foo.I);

    Assert.Equal(3, fooReflection.Get("GetI"));

    fooReflection.Set("SetI", 4);

    Assert.Equal(4, foo.I);
}

Leaving aside how much easier it make it to write the code, it also means that I now have an extension point to replace how it works if I need to. Nice.

Security

This appears to be based on contracts, which I already stated that I don't understand.

Serialization

This is cool, and I think that I figured out what extension point is for. Take a look at the code:

int instance = 1;
instance.Serialization().Binary(new MemoryStream());
instance.Serialization().Xml(new MemoryStream());

This is a general pattern of usage in the code, you have an extension method that return an ExtensionPoint<T>, on top of which you can add additional extension methods, and build up a nice API in this fashion. It also make this big library much more discoverable.

Sources

A way to get values, not sure what it is for, however.

Threading

This give us a way to handle locking in a really nice fashion, as we have already seen with SyncronizedDictionary. Just create a SynchronizableLock<T> and start calling Lock.Read(lambda) and Lock.Write(lambda).

I really like this.

Validation

This is bare bone at the moment, but it is looking really nice. Check this out:

[Fact]
public void String_NotNullOrEmpty_WithNull()
{
	string value = null;

	var ex = Assert.Throws<ArgumentNullException>(() => value.Validation().NotNullOrEmpty("value"));
	Assert.Equal("value", ex.ParamName);
}

Values

Again, this is making heavy use of messages and and I am pretty sure that I don't understand the intended usage. Digging through the code I found some interesting ideas about registering for disposal, which looks interesting. Basically, you can do the following inside classes that inherit from container:

Disposable.Add(new MyLock());

And it will be disposed when the container is disposed. Nice.

Web

Contains implementations of sources on session and web principals. I don't get sources, so I am not sure what I can make of this.

Summary

This has been a brief overview, but no means have I gone through the whole thing. Umbrella is big. At first, extension points are very strange, but they make sense once you realize how Umbrella is architected. They allow to extend a type and preserve the original value nicely.

There are some things that I really like, (Reflection, SyncronizedDictionary, Lock.Write) and some I can't figure out (sources, values, etc). I strongly suggest reading through the code, if only to get some ideas about patterns that are useful for C# 3.0.

Yesterday I reviewed Mass Transit itself, unfortunately, I missed the samples folder, which means that I didn't get critical information about how to use Mass Transit. I am going to do the same review for the samples, trying to get a feeling for how using Mass Transit feels.

Audit

This seems to be only partially done, it apparently should show how we can fire off events using Mass Transit. Again, I am impress by how simple the Cunsumes<TMsg>.All makes the code, but there is nothing there as of now.

Heavy Load

Load testing, not something that I am particularly interested at right this moment.

Publish Subscribe

This example more or less show how you can use Pub / Sub. Nothing really interesting there, but it has cleared my mind about a few things. Specifically, I was bothered by the tie of the service bus to a specific end point. What I missed was that this was the entry point end point, and that communication can (and does) flow to other end points. Once I grasped that, it became much easier to understand what was going on there.

What really bothers me, however, is this:

<component id="serviceBus"
		   lifestyle="singleton"
		   service="MassTransit.ServiceBus.IServiceBus, MassTransit.ServiceBus"
		   type="MassTransit.ServiceBus.ServiceBus, MassTransit.ServiceBus">
	<parameters>
		<endpointToListenOn>${serverEndpoint}</endpointToListenOn>
		<!-- Setter Injection -->
		<SubscriptionCache>${subscriptionCache}</SubscriptionCache>
	</parameters>
</component>
<component id="serverEndpoint"
		   lifestyle="singleton"
		   service="MassTransit.ServiceBus.IEndpoint, MassTransit.ServiceBus"
		   type="MassTransit.ServiceBus.MSMQ.MsmqEndpoint, MassTransit.ServiceBus.MSMQ">
	<parameters>
		<uriString>msmq://localhost/test_server</uriString>
	</parameters>
</component>

That sound you just heard was me, fleeing in terror. The idea that I would need to write so much XML is abhorrent to me. My idea syntax would be something like this:

facility MassTransitFacility, startable = true:
	standardCustomersServiceBus = "msmq://localhost/standard_customers"
	enterpriseCustomersServiceBus = "msmq://localhost/enterprise_customers"

extend OrdersController:
	standardCustomerPublisher = @standardCustomersServiceBus
	enterpriseCustomersPublisher = @enterpriseCustomersServiceBus

And this will allow us to define both service bus and the end points in the container, and start them as soon as they are ready. No need to do anything in the actual application code beyond just creating the container.

Web Request Reply

This is very interesting, because it shows how you can handle request reply scenarios, which are very common in data handling scenarios. In this case, the code sample show an async request / reply, using MonoRail's async actions:

public IAsyncResult BeginAsync(string requestText)
{
	_request = _bus.Request()
		.From(this)
		.WithCallback(ControllerContext.Async.Callback, ControllerContext.Async.State)
		.Send(new RequestMessage(CorrelationId, requestText));

	return _request;
}

public void EndAsync()
{
	PropertyBag.Add("responseText", msg.Text + " (and my response)");
	RenderView("Default");
}

public void Consume(ResponseMessage message)
{
	msg = message;
	_request.Complete();
}

I find it very nice. I did wonder at first how the standard EndSend() is involved here, but then I realized that it is not. It doesn't make sense for message passing semantics. When you pass the this pointer to the From() method, it will register that for accepting messages from the bus. I do wonder about such things as timeouts and memory leaks. From my observation, a failure in handling the response of a message would leak the handler instance.

To conclude, after reviewing the samples, I feel that I have a much better understanding of Mass Transit, which is good. It hasn't shifted my thinking, the way reading NServiceBus did, but that is probably because I have already read NServiceBus code.

A while ago I took a look at NServiceBus and its Distributor, after catching this post talking about Mass Transit, I decided that I really need to take an additional look at this project. Mass Transit is very similar in purpose to NServiceBus, and it uses a very similar approach. However, it has a radically different style. A lot of the concepts are shared between both projects.

As usual, I am going to post here my thoughts as I read through the code.

Opening the project is a bit overwhelming, although not as much as with NServiceBus. We have 22 projects (NServiceBus had 47 when I checked that). I tried to think how to express my initial impression, and finally settled on using the old adage about a picture worth a thousand words.

Looks like there is a lot of interesting concepts there. One thing that really stand out is that 8 of those projects are test projects. That make me feel much better about the project.

Build

The build projects handle creating the appropriate queues, there is a single class in all three projects, so I am will just ignore them.

Dashboard

The dashboard application is using MonoRail, Windsor & Brail. This make me happy, even though the dashboard is obviously in its initial stages. One point that I dislike is the use of XML configuration for Windsor.

Deferment Service

I had to look up in the dictionary to figure out what deferment means, but take a single look at the actual code made it obvious:

public interface IDefermentService
{
    int Defer(object msg, TimeSpan amountOfTimeToDefer);
}

There is no usable implementation for this, and there aren't enough tests to make it possible to understand how this is supposed to be used, either.

So far, nothing really interesting, but I am used to getting mixed results when following the Top to Bottom Review style that I like.

Host

The host is a window service that can host Mass Transit. In essence, it takes a configurator, which is an object that can return hosted services, and use that to start itself up, spinning up those services. A hosted service is a way to register and unregister to incoming messages.

There are a lot of things there that I am vague about. There is a lot of code to deal with arguments, argument maps, etc. I am not sure what it is doing there, and what it is trying to do. But I haven't read the tests yet.

The tests contains this sample, which I am not sure that I like. It looks like the service is dealing with too many things, from subscribing to messages to hosting the bus itself through message handling. This also cause my thread safety spider sense to tingle quite heavily.

I am not really interested in the configuration, from a very cursory glance, it seems to me that I would move all of that to the container and deal with it there.

Patterns

I am always suspicious when I am seeing patterns as a topic of itself. It reminds me of the application that reach GoF-complete status by implementing all of the GoF patterns. That said, I really should look inside before talking.

After reading the code, I can say that there isn't a lot of useful code there. It seems to be intended to be a repository for implementing patterns for Enterprise Integration Patterns. Something that I find very interesting is this piece of code:

public class HeartbeatMonitor :
	Consumes<Heartbeat>.All
{
	private IServiceBus _bus;

	public HeartbeatMonitor(IServiceBus bus)
	{
		_bus = bus;
	}

	public void Consume(Heartbeat message)
	{
		// do something with the heartbeat
	}
}

The Consumes<Heartbeat>.All is a really nice tagging idea. It wouldn't have occur to me, but it is a very natural way to specify what is going on there.

Subscriptions

Subscriptions in service buses play a critical role in ensuring that message would reach their destination as they should.

What appears to be a default implementation is a subscription service that is backed by a database to store those subscriptions. This looks like a really simple way of handling that, but it is not really interesting.

What is interesting is the distributed subscription cache, it is using Memcached in order to store the information. There are some issues there that bother me, however. There is minor issues with multi threaded access to that may cause issue, mainly with dictionaries used to store state. What really bothers me, however, is that a cache is not a good place to store information. By definition you can't rely on it staying there.

I haven't seen so far how the persistent storage get to update the cache. I think that I am missing something here. I assume that there are some messages flying around here that I am not seeing.

Yes, there is, and it is in the SubscriptionService, which I haven't looked deeply at yet.

I really like the way the tests are organized, by the way.

Transports

Mass Transit supports MSMQ and in the process of adding support to ActiveMQ. It is interesting to compare the two, but before I do that, I want to take a look at the MSMQ implementation. There isn't anything interesting with the Send implementation, but the Receive is interesting.

It is using GetMessageEnumerator2() to go through the messages and decide if there is any message that is worth dispatching. NServiceBus use a Peek & Read approach, instead, and then handle the message internally. There is a potential resource leak there, because the enumerator isn't using a using statement to free the enumerator in the case of exception.

The ActiveMQ implementation has just been started, from the looks of things, but just from the short code sample that there is there I can tell that it is smells heavily of Java. This is natural, since ActiveMQ is written in Java, but it is amusing to be able to recognize the style.

Service Bus

Finally! There are things to be said against reading top to bottom when the last project contains all the interesting meat. So far, I have only dealt with various infrastructure stuff, not really interesting, to tell you the truth.  Let us dive into the code...

There are formatters, which serialize an message instance to the wire. I was amused to find out that Mass Transit supports binary, XML and, of all things, JSON. I am not sure who is going to consume that, but I hope it is not running in a browser.

It looks like the Health Monitoring that appeared in the patterns section has been merged to the main line. Now we start to see how this is used. Here is an example of a service that just send a healthy heartbeat:

public class HealthClient : IHostedService
{
	readonly IServiceBus _bus;
	readonly Timer _timer = new Timer(3000);

	public HealthClient(IServiceBus bus)
	{
		_bus = bus;
		_timer.Elapsed += delegate
		{
			_bus.Publish(new Heartbeat(3, _bus.Endpoint.Uri));
		};
	}

	public void Start()
	{
		_timer.Start();
	}

	public void Stop()
	{
		_timer.Stop();
	}

	public void Dispose()
	{
		_timer.Dispose();
	}
}

I like this. There isn't a single line that I would consider wasted.

However, there are a few things that concerns me about this sample. First and foremost, the bus seems to be tied directly to a specific end point. I am pretty sure that I don't like this. The problem here is that this means that I now need to configure a bus per end point, and I find myself reluctant to do that. But let us deal with this issue later, in the mean time, I want to read the rest of the health monitoring part.

Having done that, I am not sure that I like what I am seeing. Let us take a look at the other side of the health monitoring:

public class HealthService : IHostedService
{
    private readonly IServiceBus _bus;

    public HealthService(IServiceBus bus)
    {
        _bus = bus;
    }

    public void Start()
    {
        _bus.AddComponent<HeartbeatMonitor>();
        _bus.AddComponent<Investigator>();
        _bus.AddComponent<Reporter>();
    }

    public void Stop()
    {
        _bus.RemoveComponent<Reporter>();
        _bus.RemoveComponent<Investigator>();
        _bus.RemoveComponent<HeartbeatMonitor>();
    }
}

This is the heart of what is bothering me. Why do we have to register and unregister those components? A component is a term from IoC containers, not from service buses. Using this approach give you much more control over what is going on (you can shut down services very easily this way), but I dislike it. I would route message handlers through the container and be done with it. If I wanted to dynamic start / shutdown of services, it is easy enough to do without this.

The way Mass Transit handles subscriptions is interesting. And I strongly suggest taking a look. When the service start working, it ask to get all the current subscriptions, and hold it locally. There are provisions in place to handle subscription updates, but they haven't been implemented yet.

Moving on from subscriptions, probably the most beautiful part of Mass Transit is this:

public class Consumes<TMessage> where TMessage : class
{
	public interface All
	{
		void Consume(TMessage message);
	}

	public interface For<TCorrelationId> : All, CorrelatedBy<TCorrelationId>
	{
	}

	public interface Selected : All
	{
		bool Accept(TMessage message);
	}
}

Comparing the syntax that this enable vs. the syntax that NServiceBus need... I really like this. And I am very annoyed that I have not thought about this myself.

And not it is 05:38 AM my time, and while the Mass Transit codebase is interesting, is isn't quite that much of a page turner.

I took a look at the Service Bus implementation, and found more or less what I expected. Especially frustrating is that I can see that there are a lot of features that are not used in Mass Transit itself, they are there to be exposed to users of the library.

The problem with that is that there is no reference for seeing how they are used. A sample application or two to show off what it is doing and how it is doing it would be most welcome. Especially since it would give better sense for how this works as a whole.

Update: I am an idiot, there are samples, I just missed them.

I am still unclear on the ServiceBus / Endpoint one to one mapping. I don't like this as it stand, but I am missing something. Mass Transit is 0.1, and it shows. It looks promising, and there are some interesting ideas there that I would like to see followed up.

I would like to see a service bus that doesn't try to be a container as well (NServiceBus & Mass Transit both share this fault to some extent), but rather build on top of the container and just provided the pub / sub services.

Sigh, I really don't want to have to write one.