Reactive Extensibility of Lokad.CQRS App Engine

Lokad.CQRSv2, as a application engine, uses (when it will be released) it's own internal message pipe to report system events. This works well for better testing of this project.

This is just the tip of the iceberg, though. Another use for this functionality, is that it allowed us to make the engine more extensible and yet, more simple.

For example, is no poison or failure reporting in Lokad.CQRSv2. Yeah, this sounds strange, but is based on experience with multiple cloud projects. Each project deployment tends to have highly specific needs, which depend on the situation, development context, frameworks used and team skills. For instance, sometimes we log all failures to SQL Azure, in the other cases, there is logging to blob or email sending.

It all depends.

First option (partially implemented in Lokad.CQRS v1) was to have configurable blob loggers for failures, along with queue listener for the poisons and some interceptors. This added extra code and was messy.

In the current version we just taught Lokad.CQRS to publish system events to an observable stream. By default this event stream is handled by the synchronous Trace listener (which just writes events to the trace). However, you can add your own event observer with custom event filtering and scheduling logic. Observers, observables and all the Linq-to-events syntax is provided by Microsoft Reactive extensions.

please note the difference between Domain Events and System Events. The former represent messages which are produced and handled by the user code, while application engine just routes them. System Events are produced by the application engine itself (and are used within the infrastructure).

Here's one sample snippet:

// define simple async observer  (class from Rx framework)
var events = new Subject<ISystemEvent>(Scheduler.TaskPool);

// even though Lokad.Cqrs does not need Rx lib, it supports IObservale.
builder.EnlistObserver(events);

// we don't have log, but we have plenty of events
// and all the Rx operators, like BufferWithTimeOrCount,
// that let's us avoid chatty roundtrips to the storage, if system goes wild
events
    .OfType<EnvelopeDispatchFailed>()
    .BufferWithTimeOrCount(TimeSpan.FromSeconds(1), 10)
    .Subscribe(li => SaveFailedMessagesToBlob(li));

Or let us say, we want to trace cases, where there are more than 10 quarantined per minute on a specific partition. We don't want to have more than 1 notification per hour.

This is a real requirement that I would love to have around in complex distributed systems. Fortunately, with the abundance of Rx operators, you should be able to do that in a simple manner.

events
    .OfType<EnvelopeQuarantined>()
    .WindowWithTime(TimeSpan.FromMinutes(1))
    .SelectMany(x => x.Count())
    .Where(i => i > 10)
    .Throttle(TimeSpan.FromHours(1))
    .Subscribe(c => Trace.WriteLine("More than 10 quarantined messages per minute: " + c));

Obviously, tracing is the simplest solution. In real world deployment you would probably want to do something smarter - send an email, send a message to the monitoring partition, drop SMS or whatever.

If you want to extend behavior of Lokad.CQRS bus, there is an abundance of system events to benefit from. Here are just a few:

• HostStarted
• EngineStopped
• EnvelopeSent
• EnvelopeDispatchFailed
• EnvelopeQuarantined

This creates a nice future extensibility point for Lokad.CQRS:

• Remoting system events from multiple partitions to a single trace monitor in real-time;
• Defining custom system queries with Rx (or even Qbservable) to create in-memory performance views;
• Ability to connect to any single partition and see the system log (history + real-time);

But that's something for the later.