create an async method that wraps subscribe and publish on a bus

Question

I have a kind of bus that implements this interface:

public interface IBus
{
    void Publish<T>(T t);
    void Subscribe<T>(Guid subscriptionId, Action<T> action);
    void Unsubscribe<T>(Guid subscriptionId);
}

Here is an example on how I use it:

public void PrintName()
{
    IBus bus = new Bus();
    var id = Guid.NewGuid();
    bus.Subscribe<ReplyUserName>(id, replyUserName =>
    {
        bus.Unsubscribe<ReplyUserName>(id);
        Console.WriteLine(replyUserName.UserName);
    });

    Bus.Publish(new RequestUserName());
}

And here are the RequestUserName and ReplyUserName classes:

public class RequestUserName {}

public class ReplyUserName
{
    public string UserName { get; set; }
}

However I would like to write an extension method that would wrap this with async:

public static class BusExtension
{
    public static async Task<TResult> Request<TRequest, TResult>(this IBus bus, TRequest request)
    {
        // TODO...
    }
}

So that I will be able to write the previous code in such a way:

public async void PrintName()
{
    IBus bus = new Bus();
    var replyUserName = await bus.Request<RequestUserName, ReplyUserName>(new RequestUserName());
    Console.WriteLine(replyUserName.UserName);
}

what should I write instead of the TODO?

Answer 1

You can use TaskCompletionSource<T> to wrap anything into an await-compatible method.

public static Task<TResult> Request<TRequest, TResult>(this IBus bus, TRequest request)
{
  var tcs = new TaskCompletionSource<TResult>();
  var id = Guid.NewGuid();
  bus.Subscribe<TResult>(id, result =>
  {
    bus.Unsubscribe<TResult>(id);
    tcs.TrySetResult(result);
  });
  bus.Publish(request);
  return tcs.Task;
}

Note, however, that you should ensure that the task is completed. If there's any chance that the bus won't respond to the request, you should have a timer or something that faults the TaskCompletionSource.

Answer 2

You could implement this as follows:

var taskCompletionSource = new TaskCompletionSource<TResult>();
bus.Subscribe<TResult>(id, result =>
{
    bus.Unsubscribe<TResult>(id);
    taskCompletionSource.SetResult(result);
});
bus.Publish(request);
return taskCompletionSource.Task;

You might also want to check out Reactive Extensions (Rx) as your IBus interface looks similar to the ISubject interface (http://msdn.microsoft.com/en-us/library/hh211669.aspx). The Reactive Extensions library already provides convenient extension methods similar to the one you are attempting to implement.