How to publish two messages of the same type to different worker instances based on the message content without using Send and RequestAddress?

user_5701390 0
Viewed 195

How to publish two messages of the same type to different worker instances based on the message content without using Send and RequestAddress?

My scenario is:

I am using Azure ServiceBus and Azure StorageTables.

I am running two different instances of the same worker service workera and workerb. I need workera and workerb to both consume messages of type Command based on the value of Command.WorkerPrefix.

the Command type looks like:

{
    WorkerPrefix = "a"
}

Additional constraints:

For all the messaging entities that the worker service will consume I want to add a distinct prefix whether its workera or workerb so I use the code cfg.SetEndpointNameFormatter(KebabCaseEndpointNameFormatter(args.[0], false)) where args.[0] is either "a" or "b"

Right now I only have one consumer defined that consumes a message of type Command

This gives me a topology like:

Queues:
- a-command
- b-command

Topics:
- command
  Subscriptions:
  - a-command -> fwd a-command
  - b-command -> fwd b-command

Ideally the topology would instead be:

Queues:
- a-command
- b-command

Topics:
- a-command
  Subscriptions:
  - a-command -> fwd a-command
- b-command
  Subscriptions:
  - b-command -> fwd b-command

The reason for this desired topology is such that workera would not have permissions to create subscriptions on workerb's topics and vice versa. If this is not possible I can probably resolve the permissioning issue at deployment time by creating the topology before the services start running, but would prefer the topology above.

Now from an orchestrator, either a Saga or a Future, I want to Publish two Requests of type Command. One destined for workera and one destined for workerb.

Is there anyway to configure a IPublishEndpoint based on the content of the message such that the publisher can inspect the message see that WorkerPrefix is "a" then publish to the a-command topic?

azureservicebus masstransit
1 Answers

Using MassTransit with Azure Service Bus, I would suggest taking the message routing burden away from the publisher, and moving it to the consumer. By configuring the receive endpoint and using a subscription filter each instance would add its own subscription and use a message header to filter published messages.

On the publisher, a message header would be added:

await publishEndpoint.Publish(new Command(), x => x.Headers.Set("WorkerId", "A"));

The message would be published to the Command topic, which is then routed through the subscriptions to the receive endpoints.

Using Subscription Filters

To configure the receive endpoint for instance A, whether manually configuring the receive endpoint or using a consumer definition, the receive endpoint configurator would be configured as shown below. Since the question seems to be using ConfigureEndpoints, a consumer definition approach is used.

Note that Microsoft.Extensions.Options is used to map configuration to the container, so that it is available when configuring the bus.

services.AddMassTransit(x =>
{
    x.AddConsumer<CommandConsumer, CommandConsumerDefinition>();

    x.SetEndpointNameFormatter(new KebabCaseEndpointNameFormatter("A", false));

    x.UsingAzureServiceBus((context, cfg) =>
    {
        var hostOptions = context.GetRequiredService<IOptions<Host>>();

        cfg.Host(hostOptions.Value.ConnectionString);

        cfg.ConfigureEndpoints(context);
    });
});

The consumer definition would then configure the receive endpoint:

public class CommandConsumerDefinition :
    ConsumerDefinition<CommandConsumer>
{
    readonly string _workerId;
    readonly IEndpointNameFormatter _endpointNameFormatter;

    // using IOptions here, but somehow in the container the worker settings need to be resolved
    // to get the workerId
    public CommandConsumerDefinition(IOptions<Worker> options, IEndpointNameFormatter endpointNameFormatter)
    {
        _workerId = options.Value.WorkerId;
        _endpointNameFormatter = endpointNameFormatter;
    }

    protected override void ConfigureConsumer(IReceiveEndpointConfigurator endpointConfigurator,
        IConsumerConfigurator<CommandConsumer> consumerConfigurator)
    {
        if (endpointConfigurator is IServiceBusReceiveEndpointConfigurator sb)
        {
            sb.ConfigureConsumeTopology = false;

            var subscriptionName = _endpointNameFormatter.Consumer<CommandConsumer>();

            sb.Subscribe<Command>(subscriptionName, 
                s => s.Filter = new SqlFilter($"WorkerId = '{_workerId}'"));
        }
    }
}

With the subscription filter, the header value added when the message is published is used to filter messages so that only messages matching the filter are forwarded to the worker queue.

The resulting topology:

Queues:
- a-command
- b-command

Topics:
- command
  Subscriptions:
  - a-command (WorkerId = "A") -> fwd a-command
  - b-command (WorkerId = "B") -> fwd b-command

Why not just send?

If you didn't want to deal with topics at all, you could just send to a queue and format the name yourself with the prefix. Then, just use ISendEndpointProvider (or .Send in the saga) to the destination address formatted using the endpoint name formatter combined with the prefix.

var destinationAddress = $"queue:{workerId}-{KebabCaseEndpointNameFormatter.Instance.Consumer<CommandConsumer>()}";
var endpoint = await sendEndpointProvider.GetSendEndpoint(new Uri(destinationAddress));
await endpoint.Send(new Command());

So, a couple of options that might work for you.

user_18820
Related