Rebus as a data stream

[vilinski]

Hi all,

I'm evaluating the message bus libs to replace our in-house implementation. We have 3 kinds of data. 2 of them are perfectly covered by one Rebus bus instance

• commands - direct messages
• events - the same using rebus topics

But the third one, the data, is a little bit different. It is subscribed by data key, which is a part of the message and has performance requirements. So my current solution is to create a second bus. They should use the same transport type (rabbitmq with different exchange names)

1. Is the rebus suitable for creating a data stream in a performant enough manner?
2. How or whether it is possible to create an efficient content-based routing, instead of type based? Should I implement or maybe decorate the IRouter to create topics from property values, and not from type names? There is only one message type in this case.
3. From my understanding, I have to use IBus to publish and subscribe to all message types - events, commands, and data. I found out how I could change the rabbitmq exchange and queue names to make a difference between the two buses. But I can't figure out how the routing should work in case of multiple buses. How does the rebus decide, which bus to send to or receive from? Can they have a different routers?

[vilinski]

found the answers myself. Correct me if I'm wrong

1. the performance and throughput are not a focus of Rebus here too. Each message passes at least 5 standard steps, requires a dictionary with header values, etc. The throughput is to be tested, I could experiment with removing some of the steps from the pipeline
2. content-based routing requires too many adjustments or is impossible. Other parts like serialization also rely on type-based routing. Solved using publishing to a topic, created from message content:

bus.Advanced.Topics.Publish(message.Key.ToTopicName(), message)

3. It was my mistake to assume that one of IBus instances is selected implicitly on the message type or like this. You should explicitly select one using IBusRegistry. Combined with hiding the topic internals it looks like this::

public class DataPipe(IBusRegistry busRegistry) : IDataPipe
{
    private IBus_bus = busRegistry.GetBus("data").Advanced.Topics;

    public Task Publish(ProcessData data, IDictionary<string, string>? headers = default) => 
        _bus.Publish(data.Key.ToTopicName(), data, headers);

    public Task Subscribe(DataKey key) =>
        _bus.Subscribe(key.ToTopicName());

    public Task Unsubscribe(DataKey key) =>
        _bus.Unsubscribe(key.ToTopicName());
}