Communication between modules within monolithic application (single process)

[dario-l]

How to properly achieve communication between modules which resides in the same application (single process).

Each module can publish messages. Each module can receive messages. Strictly speaking I need pub/sub.
Each module can have one or more subscribers where some of them may handle the same type of message.

How should I configure producers/subscribers to handle the same type of message by different subscribers independently (in parallel - where the business code will be executed in different/independent transaction)?

[yang-xiaodong]

I think MediatR is what you need, it does not need storage and transport

[dario-l]

Not even close. I know MediatR and look's like You do not know it. 😉 It's a different thing. Works only in-memory and do not guarantee durable messages, separation of transactions between modules, etc.

Maybe my question was not clear enough? 😞

I want to only independent subscribers on the same type of message. 😃

[yang-xiaodong]

I don’t quite understand the module you are talking about. Everyone may understand it differently. In fact, there is no strong type of messages in CAP, we use names to distinguish.

Is the consumer group what you need?
https://github.com/dotnetcore/CAP#subscribe-group

Or the following is also supported

[CapSubscribe("aaa.aaa")]
[CapSubscribe("aaa.bbb")]
[CapSubscribe("aaa.ccc")]
public void Consumer(JsonElement ele){

}

[dario-l]

Application module - I mean Bounded Context (Sales, Shipping, etc.)

Thanks for the tip. 👍🏻 I'm looking now at consumer group and looks like this could be it.
Can You explain how it work underneath with RabbitMQ?

[yang-xiaodong]

If consumers are in different groups, different queues will be created and bound to exchange with the topic type.

In other words, we use group name to create a queue with the same name and associate the routing key to the topic.

[dario-l]

That's sounds good. Thanks 😃

Two questions more:

• I there any possibility to configure subscribers through API - not by attributes?
• What is the CAP.Received table used for?

[yang-xiaodong]

I there any possibility to configure subscribers through API - not by attributes?

We don't directly support this feature, in fact, you can do this by customizing IConsumerServiceSelector.

I don't know what your purpose is. Maybe I think you don't need to do this. If your different modules want to publish and subscribe, you can add an assembly and the CAP will automatically register it.

  services.AddCap(x =>
  {
      //***
  }).AddSubscriberAssembly( ... );

What is the CAP.Received table used for?

see https://cap.dotnetcore.xyz/user-guide/en/cap/messaging/#storage

[dario-l]

I've tried to run sample Sample.RabbitMQ.SqlServer but without any luck. So I created my own sample app and everything works great. Grouping subscribers is what I was looking for. Thanks.

One thing bothers me. Is there any chance to set ConsumerThreadCount per consumer?

[dario-l]

I did some tests using RabbitMQ. After running I see two different queues in RabbitMQ. Each queue has own subscriber defined.

Unfortunately if handler from G1 was first executed then second handler waits.
I see that in debug console logs.

Setting ConsumerThreadCount=2 changes nothing.

What happened? 😞

// First handler
[CapSubscribe(Group = "G1")]
public async Task Handle(SomeMessage message) 
{
    await Task.Delay(10000);
}

// Second handler in different group
[CapSubscribe(Group = "G2")]
public async Task Handle(SomeMessage message) 
{
    await Task.CompletedTask;
}

[yang-xiaodong]

They are linearly placed in memory to ensure the order, and then use multiple processors for consumption

CAP/src/DotNetCore.CAP/Processor/IDispatcher.Default.cs

Lines 67 to 68 in b194d60

	Task.WhenAll(Enumerable.Range(0, _options.ConsumerThreadCount)
	.Select(_ => Task.Factory.StartNew(() => Processing(stoppingToken), stoppingToken, TaskCreationOptions.LongRunning, TaskScheduler.Default)).ToArray());

[NonAction]
[CapSubscribe("sample.rabbitmq.mysql")]
public void Subscriber(DateTime p)
{
    Console.WriteLine("First Start");
    Task.Delay(3000).Wait();            
    Console.WriteLine($@"{DateTime.Now} Subscriber invoked, Info: {p}");
    Console.WriteLine("First End");
}

[NonAction]
[CapSubscribe("sample.rabbitmq.mysql", Group = "group.test2")]
public void Subscriber2(DateTime p)
{
    Console.WriteLine("Second Start");
    Task.Delay(3000).Wait();
   
    Console.WriteLine($@"{DateTime.Now} Subscriber invoked, Info: {p}");
    Console.WriteLine("Second End");
}

[NonAction]
[CapSubscribe("sample.rabbitmq.mysql", Group = "group.test3")]
public void Subscriber3(DateTime p)
{
    Console.WriteLine("Third Start");
    Task.Delay(3000).Wait();
    Console.WriteLine($@"{DateTime.Now} Subscriber invoked, Info: {p}");
    Console.WriteLine("Third End");
}

Command line output:

Third Start
Second Start
First Start
2021/10/29 22:28:25 Subscriber invoked, Info: 2021/10/29 22:28:22
First End
2021/10/29 22:28:25 Subscriber invoked, Info: 2021/10/29 22:28:22
Third End
2021/10/29 22:28:25 Subscriber invoked, Info: 2021/10/29 22:28:22
Second End

[dario-l]

@yang-xiaodong So what's the point of creating multiple queues (per group)?

[yang-xiaodong]

So what's the point of creating multiple queues (per group)?

Designed in this way to support broadcasting, not alltransports are easy to achieve this

[dario-l]

@yang-xiaodong Looks like now I have:

• ConsumerThreadCount = 1 - processing up to 1 message at time
• ConsumerThreadCount = 2 - processing up to 4 messages at time

Once again. Each Group should have own processing pipeline (channel).

If I want to everything be linear that I should only use 1 (default) group.

• ConsumerThreadCount>1 - should increase threads on each processing pipelines
• ConsumerThreadCount=1 - each processing pipeline (each group) is linear

[yang-xiaodong]

I think this is just to establish more connections to the Broker server in order to receive messages faster, just like starting multiple instances

[dario-l]

@yang-xiaodong One connection is fine or maybe I'm something do not understand.
Can you tell me how many subscribers will be created by CAP in your example described above?