Publishing messages into memory transport are blocking until execution is done

[rgbweb]

Hi, thank you for providing this nice to start and easy to use message bus library. I searched a while for a message bus with quite exactly these features and simpleness. For one of my projects, I plan to start with a simple self-contained application running in a single process, where all parts are connected through an in-memory message bus. But I want to have the ability to scale out to multiple processes/servers without having to rewrite the whole application. This plan looks to be easily made by using your library.

I have a question regarding the pub/sub behavior of the MemoryMessageBus. For my understanding the bub/sub messaging using the Publish-method should work like a queue where I put things in and stop worrying how long the execution takes. On the other hand, the request-response messaging using the Send-method with the await keyword should block the execution of my calling method until the response has arrived. If I see it correctly, this behavior is supported by all available transport implementations but not by the MemoryMessageBus. The MemoryMessageBus awaits the execution of the work even if I use the Publish-method. I did not see a note about it in the docs but there is an example in the DomainEvents sample projects. The OrderSubmittedHandler returns a delay of one second which blocks the Post-method in the OrdersController for that second before it can respond to the HTTP request.

Is this the intentional behavior? And what is the reason for it? Maybe I simply have a wrong understanding about the intended behavior of general pub/sub messaging in the message bus.

Cheers from Germany

[zarusz]

Hello @rgbweb, yes you got it right. That is the behavior with the current implementation of the memory provider.

Now, it was implemented intentionally for the use case where you publish a domain event and want to be sure that the consumer complete within the same unit of work (e.g. ongoing API web request) - so that the side effects complete before the request finishes (and the Publish returns).

This should be better documented, so I will make sure that soon this is the case.

Now, I do agree that in principle it would be better (and likely more expected by users) that the Publish is non-blocking as you said implemented via some in memory queue. However, this would require few breaking changes on the lifecycle of consumers and would not be backward compatible (and surely not work with the above domain events scenario). I think we could add an option to the MemoryBus where it would switch it into this new queue / non-blocking when doing Publish - the default behaviour would still work like today but you could enable the new flow.

What do you think?

[rgbweb]

Hi Tomasz, thank you for the explanations. I think the idea of having new option in the in-memory configuration would be the easiest. Another idea would be, to have a second transport package for in-process message bus with queues instead of the blocking Task-await implementation. But this could be more work and even not less confusing to users of the library.

I already played a bit around yesterday, to check if there could be a simple solution to add a queue to the MemoryMessageBus. I assumed it could be quite easy by using the System.Threading.Tasks.Dataflow NuGet package wich is already added to the Host-project of the solution. Some time ago I read a blog post (https://michaelscodingspot.com/c-job-queues-part-3-with-tpl-dataflow-and-failure-handling/) about using it as a fast in-memory queue with built-in support for pub/sub with multiple subscribers and limiting parallel execution of subscriptions. There is also a way to filter the queue messages with a predicate, which allows to deal with different "Topics" in the message bus implementation.

But I struggled a when I tried to write the implementation code for the message bus. I suspect, I do not understand enough of the internal message-flow in the base implementations and I do not have the time to install some of the transport providers to debug the existing transport implementations. I tried to use the DataFlow queue for the pub/sub and the request-response implementations. But by looking at the existing code, I also assume that some of the built-in features of System.Threading.Tasks.Dataflow are also manually implemented in SlimMessageBus. It looks like handling of parallelism and different subscriptions for a single topic can also be done by internal logic. I think it would be best to reuse the provided logic from the host instead of using similar logic of an external library.

Generally, I would really like to contribute this feature into the SlimMessageBus or at least help by implementing it. But unfortunately it looks like, I will not have much time for it in the upcoming weeks. So, I cannot say if I can do it at all.

[robert94p]

no plans to add this feature?

[zarusz]

Well, I am more than open to having this feature, just there are other areas I am evolving SMB on at the moment.

I am open to contributions from the community. Feel free to raise a PR around Memory transport.

Otherwise, I will get to this eventually after the 2.0.0 release.

[maciejrzs]

Just upvoting this feature. It would be really good to have non-blocking publishing for the in-memory message bus.

[zarusz]

I am looking at implementing this feature now. Stay tuned.

[zarusz]

You can check out the feature preview: https://www.nuget.org/packages/SlimMessageBus.Host.Memory/2.3.0-rc2

How to use and configure is described in the memory docs.

There are also other improvements around global error handlers.

Feedback welcome.

[zarusz]

It is now incorporated into the master.

Still in preview until I add a few other requested features: https://www.nuget.org/packages/SlimMessageBus.Host.Memory/2.3.0-rc3

Memory Docs

Feedback welcome.