CODEWITHSUNDEEP
springspring-cloudspring-cloud-stream
idipous
idipous

Reputation: 2910

Spring Cloud Stream topic per message for different consumers

The topology I am looking for is

enter image description here

So far I have not seen a way to define the topic per message in Cloud Stream. I understand that the consumers will be bound to specific topic but how does the producer sets the topic per message before sending the message to the exchange?

source.output().send(MessageBuilder.withPayload(myMessage).build());

Does not provide any way to set the topic for the exchange to route to the proper consumer.

Or maybe I don't understand something correctly?

UPDATE

I would expect not to receive the message in the consumer due to the bindingRoutingKey being 2222 and I am sending with routeTo 1111. But I still receive it on the consumer.

Producer Properties:

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

spring.cloud.stream.bindings.output.content-type=application/json
spring.cloud.stream.bindings.output.destination=messageExchange
spring.cloud.stream.rabbit.bindings.output.producer.routing-key-expression=headers['routeTo']


@EnableBinding(Source.class)
@SpringBootApplication
public class Application {

   public static void main(String[] args) {
    SpringApplication.run(Application.class, args);
  }

}

Sender:

source.output().send(MessageBuilder.withPayload(mo).setHeader("routeTo", "1111").build());

And the Consumer:

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

spring.cloud.stream.bindings.input.destination=messageExchange
spring.cloud.stream.rabbit.bindings.input.consumer.bindingRoutingKey=2222

Application: 

@SpringBootApplication
@EnableBinding(Sink.class)
public class Application {

private static final Logger log = LoggerFactory.getLogger(Application.class);

public static void main(String[] args) {
    SpringApplication.run(Application.class, args);
}

@StreamListener(Sink.INPUT)
public void ReceiveMo(String moDTO) {
    log.info("Message received moDTO: {}", moDTO);
}

}

SECOND UPDATE

With the suggestions in the accepted answer below. I was able to make it work. Needed to remove the exchanges and queues from RabbitMQ using its UI and restart the RabbitMQ docker image.

Upvotes: 2

Views: 1662

Answers (1)

Gary Russell
Gary Russell

Reputation: 174564

The the routingKeyExpression rabbitmq producer property.

e.g. ...producer.routing-key-expression=headers['routeTo']

then

source.output().send(MessageBuilder.withPayload(myMessage)
    .setHeader("routeTo", "Booking.new")
    .build());

Note that the destination is the exchange name. By default, the binder expects a Topic exchange. If you wish to use a Direct exchange instead, you must set the exchangeType property.

Upvotes: 5

Related Questions