Commands executed synchronously when using NServiceBus

[benfoster]

Hi guys,

Thought I would post here rather than keep spamming Darren on Twitter :)

I'm working on a CQRS demo at https://bitbucket.org/retroviz/cqrsdemo and just added NServiceBus into the mix.

Process is:

CQRSDemo.Web sends commands to SimpleCQRSDemo.CommandService
SimpleCQRSDemo.CommandService dispatches commands to SimpleCQRSDemo.Worker (via NServiceBus)
SimpleCQRSDemo.Worker is configured as both publisher and subscriber of events.

Now everything works fine except that it appears the commands are dispatched Synchronously. To test I introduced some latency into the CommandHandlers and EventHandlers. The CommandService (WCF) does not return from Execute until both CommandHandlers and EventHandlers have completed.

The net result is that the web site takes 5 seconds to process the request where I would expect it to be instant.

Any help would be appreciated.

[benfoster]

Ignore me. I was calling Execute instead of Send. School boy error!

[darrencauthon]

It's not a school boy error -- when we're missing so much documentation, people are going to get tripped up.

So on behalf of us, I apologize to you for this issue. :)

[tyronegroves]

Darren is absolutely correct I apologize for the confusion. I will try to get some help to start some documentation.

[benfoster]

You don't need to apologize. Strictly speaking I should know better as the same convention is used in NServiceBus. I hope to blog about my experience with SimpleCQRS quite soon so this may be helpful to people trying out the framework.

One thing that would be good is if this was on Nuget, especially when we are using other libraries like NServiceBus - The current NServiceBus extension is using an older version (I had to do this manually in the source).

[tyronegroves]

Ok we will work on getting everything Nugetified. Also I was looking at your sample. 1. Your command handler PostNewTweetCommandHandler can just inherit from CreateCommandHandler<> (its makes the create handlers alot easiler to test too).
2. I never though of doing all the NServiceBus processing the way you have it. We would have the command handlers in a NServiceBus endpoint and the event handlers in another NServiceBus endpoint. But very cool the way you have it first time seeing that.

[benfoster]

Cool I'll update the handler.

The main reason I've combined the two is a bit of selfish one. I'm looking at using CQRS and NServiceBus on Azure and since we pay per instance, I want to save a bit of money by combining the two :)

I will be adding some event handlers to the web project shortly. I'm just looking into SignalR, the idea being that we subscribe to a "TweetPostedEvent" in our web project and use SignalR to push it down to the client, rather than constantly polling from the client.

[benfoster]

As discussed on twitter I'm now having problems using the "Worker" being both a publisher and subscriber, when I need to to publish to the web application.

I did wonder why not just bypass SimpleCQRS and register the message/handlers manually but the issue is I am subscribing to SimpleCQRS domain events so need the messages and handlers converting into NServiceBus compatible types.

This is done for us when we use the "SubscribeToDomainEvents" extension method. Unfortunately since I can't use this and "UseNsbEventBus" I've been trying to register the message types and handlers in the same way.

This is what I tried:

var runtime = new SimpleCQRSDemoRuntime();
var config = Configure.With()
    .DefaultBuilder()
    .BinarySerializer()
    .SimpleCqrs(runtime)
        .UseLocalCommandBus()
        .UseNsbEventBus();

new ConfigEventBus().Configure(config, runtime);

This calls the following:

public void Configure(ConfigSimpleCqrs config, ISimpleCqrsRuntime runtime)
{
    Configurer = config.Configurer;
    Builder = config.Builder;

    var serviceLocator = runtime.ServiceLocator;
    var typeCatalog = serviceLocator.Resolve<ITypeCatalog>();

    var domainEventBusConfig = GetConfigSection<DomainEventBusConfig>();
    var domainEventTypes = typeCatalog.GetDerivedTypes(typeof(DomainEvent));
    var domainEventMessageTypes = new List<Type>();

    var bus = (UnicastBus)config
        .MsmqTransport()
        .UnicastBus()
            .LoadMessageHandlers(new First<DomainEventMessageHandler>())
            .CreateBus();

    RegisterAssemblyDomainEventSubscriptionMappings(domainEventBusConfig, domainEventTypes, domainEventMessageTypes, bus);
    RegisterDomainEventSubscriptionMappings(domainEventBusConfig, domainEventTypes, domainEventMessageTypes, bus);

    bus.Started += (s, e) => domainEventMessageTypes.ForEach(bus.Subscribe);
}

When I run the worker I get an NServiceBus error:

"Subscription message from cqrsdemo.web@computername arrived at this endpoint, yet this endpoint is not configured to be a publisher"

Perhaps this is because we are overriding the bus in the above method? I'm a little bit lost at this point.

Surely the entire point of using NServiceBus is that we can add multiple subscribers for integration etc.?

[tyronegroves]

I guess I am a little confused as to why you want to subscribe to receive events in the web application. In CQRS the web app (or client) does not receive events but only read from the view models and send commands.

[benfoster]

But this client is no different to any other subscriber. With the demo as it stands I have the Worker handling both commands and events. Now I decide to offload the processing of some commands onto another server for performance reasons, so I add another subscriber. Now I'm in the same position that I'm in currently.

[benfoster]

I would also add that I do see this (web app subscribing to events) being more common (if it's not already). The technology I'm trying to test (SignalR) opens up massive opportunities as we can now effectively push directly to the browser rather than resorting to polling like we do currently.

[tyronegroves]

I can look into getting that working, but I have a project that I am working on also. It took me a long time to get the NServiceBus integration working so I know its going to take a lot of thinking and time to get it working correctly.

[benfoster]

I think I'll get there eventually, just needed a few days away from the keyboard.

I've all but given up on Ncqrs now so will focus my efforts on SimpleCQRS :)

[benfoster]

Tyrone,

Really close to getting this working now. I've got the publisher subscribing to it's own events (https://bitbucket.org/retroviz/cqrsdemo/changeset/0a505c8f4304#chg-SimpleCQRS/SimpleCQRSDemo.Worker/UnicastBusExtensions.cs).

The only problem now is publishing them. When I try to publish an event I get the error:

"Cannot create an instance of an interface" at:

    public void PublishEvent(DomainEvent domainEvent)
    {
        Bus.Publish<IDomainEventMessage>(message => message.DomainEvent = domainEvent);
    }

In NsbEventBus.

Any idea what I'm missing?

[geoffreysmith]

Ben, e-mail me your project and I'll take a look at it.

[geoffreysmith]

Nevermind I see the bitbucket link.

[benfoster]

Geoffrey, just pushed up where I am so far.

The CQRSDemo.Web project is subscribed to TweetPostedEvent. I'm trying to also subscribe the worker project to the same event (so I can have both pub/sub on the same instance).

If you look in the worker EndPoint config you'll see the following code commented out at startup:

//((UnicastBus)Bus).SubscribeToDomainEvents(runtime);

This subscribes the worker to all domain events.

Now I think I've just come across something interesting.

When we publish the events to the Web project, we call NsbEventBus.PublishEvents:

    public void PublishEvents(IEnumerable<DomainEvent> domainEvents)
    {
        var domainEventMessages = domainEvents.Select(CreateDomainEventMessage).ToList();
        domainEventMessages.ForEach(message => Bus.Publish(message));
    }

However when we add the subscriptions to the worker we call NsbEventBus.CreateEvent:

    public void PublishEvent(DomainEvent domainEvent)
    {
        Bus.Publish<IDomainEventMessage>(message => message.DomainEvent = domainEvent);
    }

I'm beginning to think that the above will always fail. In the NServiceBus docs, if publishing an interface you have to call Bus.CreateInstance on it first.

This has the result of calling Activator.CreateInstance which is effectively what we are doing for the PublishEventS method.

So I'm led to believe that I either need to change PublishEvent to create the instance in a similar way or, figure out why the set up of the Worker project doesn't publish in the same way as the web project.

Desperate to get this working, if anything, so I can sleep at night :)

[tyronegroves]

I think your problem is combining the command handlers and event handlers in the same endpoint. If you split them out then you wouldn't have to subscribe to receive events to yourself.

[benfoster]

Hi Tyrone,

This is a fairly common scenario when using NServiceBus and works fine on its own (see https://bitbucket.org/retroviz/nsbpubsub).

I'm sure I'm close now. I just don't understand why the domain event is published in two different ways when both web and worker projects use the same type of event bus. For the web project the call stack is

>   SimpleCqrs.NServiceBus.DLL!SimpleCqrs.NServiceBus.Eventing.NsbEventBus.PublishEvents(System.Collections.Generic.IEnumerable<SimpleCqrs.Eventing.DomainEvent> domainEvents) Line 20  C#
    SimpleCqrs.DLL!SimpleCqrs.Domain.DomainRepository.Save(SimpleCqrs.Domain.AggregateRoot aggregateRoot) Line 42 + 0xf bytes   C#
    SimpleCqrs.DLL!SimpleCqrs.Commanding.CreateCommandHandler<SimpleCQRSDemo.Commands.PostNewTweetCommand>.Handle(SimpleCQRSDemo.Commands.PostNewTweetCommand command) Line 15 + 0xc bytes  C#

For the worker it's

>   SimpleCqrs.NServiceBus.DLL!SimpleCqrs.NServiceBus.Eventing.NsbEventBus.PublishEvent(SimpleCqrs.Eventing.DomainEvent domainEvent) Line 15    C#
SimpleCqrs.NServiceBus.DLL!SimpleCqrs.NServiceBus.Eventing.DomainEventMessageHandler.Handle(SimpleCqrs.NServiceBus.Eventing.IDomainEventMessage message) Line 19    C#

[benfoster]

I don't think I'm going to have much success with this. Having separate instances for pub/sub is not an option for me as I was hoping to get SimpleCQRS running on Azure, and I don't want to pay for two instances (one for pub, one for sub).

The problem isn't subscribing to the domain events (this was quite straightforward once I'd stepped through the SimpleCQRS.NserviceBus source), its how the events are published.

The current implementation does not expect the event handling to be done on the same instance using a NsbEventBus. Essentially PublishEvents just calls the DomainEventMessageHandler which then expects to call PublishEvent which needs to be on a LocalNsbEventBus - so in short, as you pointed out, it's just not set up to work this way.

Hopefully, I'll get chance to work with SimpleCQRS on another project. Thanks for your help anyway.

[stemarie]

Ben,

I was just curious, why are you using (in NSB) event publishing instead of just command execution?