Spring cloud stream RabbitMQ routing messages dynamically

[abhdeb]

I have implemented the example as shown here Dynamic Destinations Spring Dynamic Destination

In the rabbitmq, it is creating an exchange dynamically, but there is no option to provide binding or routing key. My requirement is to send a message to this dynamically created exchange with a routing key. How would i need to implement this to setup the routing key?

@Component public class DDProducerBean {

@Autowired
private BinderAwareChannelResolver poChannelResolver = null;

public void publish(DDSocketVO ddSocketVO) throws Exception {
    this.poChannelResolver.resolveDestination(ddSocketVO.getDestination()).send(MessageBuilder.withPayload(new ObjectMapper().
            setVisibility(PropertyAccessor.FIELD, Visibility.ANY).
            writeValueAsString(ddSocketVO)).build());
}

My requirement is to route messages dynamically at runtime based on the payload. So i need a way to set the routing key at run time. As a result i can't use producer.routingKeyExpression, which is done in config or properties or yml, and not dynamically at run time, so that won't serve the purpose.

This implementation is specific for spring.cloud.stream.dynamicDestinations where the BinderAwareChannelResolver takes care of dynamically creating/binding the outbound channel for these dynamic destinations. But it doesn't talk about how to bind the outbound channel to a queue and how to specify a routing key expression. As these are dynamic channels/ producers, we can't use producer.routingKeyExpression as the producer is created dynamically.

[garyrussell]

Producers are not responsible for binding queues; consumers are.

There is some support for a producer binding queues using the requiredGroups property, but that is for fixed destinations; there is no such support for dynamic destinations.

You can programmatically declare queues and bindings using the AmqpAdmin bean after you have resolved the destination and before sending messages.

[abhdeb]

That sounds perfect, i had the same idea, thank you.

[abhdeb]

In case of routingKeyExpression, producer is responsible for setting the expression- for example
System.setProperty("spring.cloud.stream.rabbit.bindings.websocketOutputChannel.producer.routingKeyExpression", "'adam.ir.list'"); Do you mean "Producers are not responsible for binding queues; consumers are" only incase of dynamic destination implementation?

[garyrussell]

There are two "routing keys" involved. The routing key used by the producer when sending messages and the routing key used to bind the queue. The latter can be a pattern (for a topic exchange).

The default producer routing key is the destination name. The default consumer binding routing key is # (with a topic exchange and a non-partitioned destination), meaning all consumers get a copy of the message.

With dynamic destinations the producer routing key is always the destination name; you would normally want to use # in the binding routing key.

[abhdeb]

Based on your suggestion we have now created the binding using AmqpAdmin and a routing key and it is working as expected, thank you.

Just curious as why routing key api is not available in dynamic destinations? In this scenario every time a new exchange is created and then using AmqpAdmin bind it to the defualt exchange with the routing key, which seems like a workaround. Ideally when we send a message, if we can send the routing key, then exchange can route it according without having to create new exchange for each request and then bind them using AmqpAdmin.

[garyrussell]

It just hasn't been asked for before; I guess that, typically, with dynamic routing, the consumers are already in place. We would need some kind of templating mechanism, such as something like...

...stream.rabbit.bindings.*.producer.routingKeyExpression

and

...stream.rabbit.bindings.*.producer.bindQueue
...stream.rabbit.bindings.*.producer.bindingRoutingKey

for requiredGroups for use when binding dynamic destinations.

Where * represents the dynamically created dest.

With your work around, you should keep track of which destinations you have already bound, to avoid re-binding on every request.