Whenever I’m developing a DDD web app, I’m using CQRS and Domain Events and this means I have a lot of commands and event handlers. Therefore I’m using a durable service bus which means that if the server crashes, all the unhandled messages are sent again to be handled. But I have a problem: all my commands are handled asynchronously because this is how my service bus works. And it works this way because a message can be handled anywhere in that process (usually in a background thread) or in another process if we have a distributed app.

Asynchrony means I can only ‘send’ a command but I can’t expect an answer. Basically it’s only

bus.Send(new DoSomething(){});

and this sucks when application services are implemented as command handlers. Because the user expects some feedback. If the command fails you want to return some error and this is quite tricky in this scenario. The usual solution is to do some client side server polling checking the state of the operation or waiting for a read model. This is obviously cumbersome.

However, we can apply a similar concept but on server side using a mediator. Simply put, we’ll have a an object in charge of delivering the command result to our controller. It goes like this in the controller

This is NancyFx code but it’s easy to understand, instead of returning an ActionResult or directly the model, I’m returning a NancyResponse which can be a HttpStatusCode as well. This is not important, the important part is the use of the mediator. We send the command to be handled then we ask the mediator for a listener that will be used to await the command result. If the result doesn’t arrive in a couple of seconds or if the operation is cancelled, exceptions are thrown.

In the handler we do things as usual, the only change is that now the result is sent to the mediator. And here is the mediator itself

publicclassCommandResultMediator:ICommandResultMediator{publicCommandResultMediator(){ResultCheckPeriod=100;}publicvoidAddResult<T>(GuidcmdId,Tresult)whereT:class{Listenerl=null;if(_items.TryRemove(cmdId,outl)){l.Result=result;}}/// <summary>
/// How often to check if a result has arrived, in ms.
/// Default is 100 ms
/// </summary>
publicintResultCheckPeriod{get;set;}voidRemove(GuidcmdId){Listenerl=null;_items.TryRemove(cmdId,outl);}publicintActiveListeners{get{return_items.Count;}}ConcurrentDictionary<Guid,Listener>_items=newConcurrentDictionary<Guid,Listener>();publicIResultListenerGetListener(GuidcmdId,TimeSpan?timeout=null){timeout=timeout??TimeSpan.FromSeconds(5);varlistener=newListener(timeout.Value,cmdId,this);listener.UpdatePeriod=ResultCheckPeriod;_items.TryAdd(cmdId,listener);returnlistener;}classListener:IResultListener{privatereadonlyGuid_cmdId;privatereadonlyCommandResultMediator_parent;publicListener(TimeSpantimeout,GuidcmdId,CommandResultMediatorparent){_cmdId=cmdId;_parent=parent;TimeoutTime=DateTime.Now.Add(timeout);}privateDateTimeTimeoutTime{get;set;}publicobjectResult{get;set;}publicintUpdatePeriod=100;publicasyncTask<T>GetResult<T>(CancellationTokencancel=default(CancellationToken))whereT:class{while(Result==null){if(DateTime.Now>=TimeoutTime){_parent.Remove(_cmdId);thrownewTimeoutException();}if(cancel.IsCancellationRequested){_parent.Remove(_cmdId);thrownewOperationCanceledException();}awaitTask.Delay(UpdatePeriod,cancel).ConfigureAwait(false);}returnResultasT;}}}

Btw, the mediator should be a singleton managed by your favourite DI Container.

Note that the mediator is part of my CavemanTools library so if you’re using it, or SqlFu or MvcPowertools you only need to update to the last version.

So pretty much that’s it! The easiest way to return a result from an async command handler.