Passing dynamic list of queues into a queue listener

user_10776719 0
Viewed 2270

I have a Spring Boot application that spins consumers for a section of queues and I want to be able to add queues to those consumers at run time.

I've installed the event exchange plugin (https://www.rabbitmq.com/event-exchange.html) and I created a dedicated queue that is bound to the amq.rabbitmq.event exchange. I can see the events coming in when I declare the queues statically.

How would I accomplish this run time magic? I've seen people use the property file, but I would prefer not to have to modify the property file during run time as I add more queues

@Component
public class MessageConsumer {

    List<String> allQueues = new ArrayList<String>();

    public MessageConsumer() {
        allQueues.add("queue1");
        allQueues.add("queue2");
        allQueues.add("queue3");
    }

    @RabbitListener(id = "event", queues = {"custom-emp-queue-events"}) // create this queue in rabbitmq management, bound to amqp exchange
    public void processQueueEvents(Message message) {
        ... add the queue to the allQueues list on queue.created ...
    }

    @RabbitListener(id = "process", queues = allQueues.stream().toArray(String[]::new) ) // this is where the "issue" is
    public void processMessageFromQueues(String messageAsJson) {
        ... process message ...
    }
}
spring-amqp spring-boot rabbitmq
1 Answers

This can be done with a SpEL expression over there:

@RabbitListener(id = "process", queues = "#{messageConsumer.allQueues}" )

But you have to add a public getter for that allQueues.

See more info in the Reference Manual: https://docs.spring.io/spring-amqp/docs/2.1.3.RELEASE/reference/html/_reference.html#async-annotation-driven

UPDATE

@Autowired
private RabbitListenerEndpointRegistry listenerEdnpointRegistry;

@RabbitListener(id = "event", queues = {"custom-emp-queue-events"}) // create this queue in rabbitmq management, bound to amqp exchange
public void processQueueEvents(Message message) {
    ((AbstractMessageListenerContainer) this.listenerEdnpointRegistry.getListenerContainer("process")).addQueueNames(...);
}
user_27565470
Related