Using Rx to Test CQRS App Engine Functionality

Current trunk of Lokad.CQRS does not have default logger interface which plagued all my previous projects. It had a few extension methods and looked like this:

public interface ILog
{
  void Log(LogLevel level, object data);
  void Log(LogLevel level, Exception ex, string message);
  ...
}

log.Debug("Starting process '{0}'", _processName);

Usual and boring abstraction (part of logging application block in Lokad Shared libraries).

In Lokad.CQRSv2 I've discarded ILog completely (among many other things) in favor of something more explicit:

public interface ISystemObserver
{
    void Notify(ISystemEvent @event);
}

public interface ISystemEvent { }

Where an event is just:

public sealed class QuarantinedMessage : ISystemEvent
{
    public Exception LastException { get; private set; }
    public string EnvelopeId { get; private set; }
    public string QueueName { get; private set; }

    public QuarantinedMessage(Exception lastException, string envelopeId, string queueName)
    {
        LastException = lastException;
        EnvelopeId = envelopeId;
        QueueName = queueName;
    }
}

And usage is as straightforward as it seems. Here's the bit from the code:

if (_quarantine.Accept(context, ex))
{
    _observer.Notify(new QuarantinedMessage(
        ex, 
        context.Unpacked.EnvelopeId, 
        context.QueueName));
    _inbox.AckMessage(context);
}
else
{
    _inbox.TryNotifyNack(context);
}

If this starts looking like event-driven architecture to you - you are correct. Instead of writing strings to some buffer, I send strongly-typed system events to the internal memory pipe. They are one-way and in the passed tense, just like in domain messages. And I'm perfectly fine with defining a few dozen of message classes, since this does not increase code complexity (it reduces it, actually).

Al this leads to cleaner design and a few positive side effects. For example, I can test my service bus like this:

[Test]
public void FailureEndsWithQuarantine()
{
    EnlistHandler(x =>
        {
            throw new InvalidOperationException("Fail: "+x);
        });

    Events
        .OfType<QuarantinedMessage>()
        .Subscribe(cm =>
            {
                Assert.AreEqual("Fail: try", cm.LastException.Message);
                CompleteTestAndStopEngine();
            });

    RunEngineTillStopped(() => SendString("try"));

    Assert.IsTrue(TestCompleted);
}

If you are interested in the details - just check out Lokad.CQRS project sources.

Please, be warned that the codebase is still highly unstable and will change a lot in the next few days.

PS: Sagas and complex interactions can be unit-tested in a similar manner as well, provided your engine supports some sort of fast in-memory partitions.

Comments

[Chris martin]

I've been following the repo very closely and am very excited by what I'm seeing. I can't wait for v2.

Keep it up. You're doing a great job!

[Rinat Abdullin]

Chris, thanks for the encouragement :)

Lokad.CQRSv2 should be stabilized within the next week. We'll test it in our production afterwards, making a beta release.