Spring & RabbitMQ - register queue at runtime

Question

How can I create new queue bound to Fanout exchange and run it during runtime? So far I have this:

Map<String, Object> arguments = new HashMap<>();
    arguments.put("x-message-ttl", 600000L);

    GenericBeanDefinition runtimeQueueBean = new GenericBeanDefinition();
    runtimeQueueBean.setBeanClass(Queue.class);
    runtimeQueueBean.setLazyInit(false);
    runtimeQueueBean.setAbstract(false);
    runtimeQueueBean.setAutowireCandidate(true);
    ConstructorArgumentValues queueConstrArgs = new ConstructorArgumentValues();
    queueConstrArgs.addIndexedArgumentValue(0, queueName);
    queueConstrArgs.addIndexedArgumentValue(1, true);
    queueConstrArgs.addIndexedArgumentValue(2, false);
    queueConstrArgs.addIndexedArgumentValue(3, false);
    queueConstrArgs.addIndexedArgumentValue(4, arguments);
    runtimeQueueBean.setConstructorArgumentValues(queueConstrArgs);
    this.context.registerBeanDefinition("nejm", runtimeQueueBean);


    GenericBeanDefinition runtimeFanoutExchange = new GenericBeanDefinition();
    runtimeFanoutExchange.setBeanClass(FanoutExchange.class);
    runtimeFanoutExchange.setLazyInit(false);
    runtimeFanoutExchange.setAbstract(false);
    runtimeFanoutExchange.setAutowireCandidate(true);
    ConstructorArgumentValues constructorArgumentValues = new ConstructorArgumentValues();
    constructorArgumentValues.addIndexedArgumentValue(0, "staticCache");
    runtimeFanoutExchange.setConstructorArgumentValues(constructorArgumentValues);
    this.context.registerBeanDefinition("staticCache", runtimeFanoutExchange);


    GenericBeanDefinition runtimeBinding = new GenericBeanDefinition();
    runtimeBinding.setBeanClass(Binding.class);
    runtimeBinding.setLazyInit(false);
    runtimeBinding.setAbstract(false);
    runtimeBinding.setAutowireCandidate(true);
    constructorArgumentValues = new ConstructorArgumentValues();
    constructorArgumentValues.addIndexedArgumentValue(0, queueName);
    constructorArgumentValues.addIndexedArgumentValue(1, Binding.DestinationType.QUEUE);
    constructorArgumentValues.addIndexedArgumentValue(2, "staticCache");
    constructorArgumentValues.addIndexedArgumentValue(3, "");
    runtimeBinding.setConstructorArgumentValues(constructorArgumentValues);
    this.context.registerBeanDefinition("bajnding", runtimeBinding);


    GenericBeanDefinition runtimeMessageListenerAdapter = new GenericBeanDefinition();
    runtimeMessageListenerAdapter.setBeanClass(MessageListenerAdapter.class);
    runtimeMessageListenerAdapter.setLazyInit(false);
    runtimeMessageListenerAdapter.setAbstract(false);
    runtimeMessageListenerAdapter.setAutowireCandidate(true);
    constructorArgumentValues = new ConstructorArgumentValues();
    constructorArgumentValues.addIndexedArgumentValue(0, this);
    constructorArgumentValues.addIndexedArgumentValue(1, new RuntimeBeanReference("jackson2JsonMessageConverter"));
    runtimeMessageListenerAdapter.setConstructorArgumentValues(constructorArgumentValues);
    this.context.registerBeanDefinition("mla2", runtimeMessageListenerAdapter);



    GenericBeanDefinition runtimeContainerExchange = new GenericBeanDefinition();
    runtimeContainerExchange.setBeanClass(SimpleMessageListenerContainer.class);
    runtimeContainerExchange.setLazyInit(false);
    runtimeContainerExchange.setAbstract(false);
    runtimeContainerExchange.setAutowireCandidate(true);
    MutablePropertyValues propertyValues = new MutablePropertyValues();
    propertyValues.addPropertyValue("connectionFactory", new RuntimeBeanReference("connectionFactory"));
    propertyValues.addPropertyValue("queues", new RuntimeBeanReference("nejm"));
    propertyValues.addPropertyValue("messageListener", new RuntimeBeanReference("mla2"));
    runtimeContainerExchange.setPropertyValues(propertyValues);
    this.context.registerBeanDefinition("defqueue", runtimeContainerExchange);

The problem is that queue/exchange is not created at the runtime, and I have to manually start the listener (unless I call this.context.start() - but I don't know if this is correct approach).

My question - is there some way to magically start all generated beans in runtime (something like this.context.refresh() - this exists but doesn't work or similar)?

UPDATE:

This is how I do it currently (this approach works, but don't know if correct one)

    Map<String, Object> arguments = new HashMap<>();
    arguments.put("x-message-ttl", 600000L);
    Queue queue = new Queue(queueName, true, false, false, arguments);

    FanoutExchange exchange = new FanoutExchange("staticCache");

    Binding binding = new Binding(queueName, Binding.DestinationType.QUEUE, "staticCache", "", null);

    rabbitAdmin.declareQueue(queue);
    rabbitAdmin.declareExchange(exchange);
    rabbitAdmin.declareBinding(binding);

    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
    container.setConnectionFactory(this.connectionFactory);
    container.setQueues(queue);
    container.setMessageListener(new MessageListenerAdapter(this, this.converter));

    container.start();

Answer 1

You can't do that way. BeanDefinition and this.context.registerBeanDefinition are for parsing phase of your application context lifecycle.

If you app is already there, the application context won't accepts any BeanDefinition.

Yes, you can declare Queue and its Binding to the exchange manually at runtime. And also you even can create SimpleMessageListenerContainer manually and make it worked.

And what is good for you that you just need to use their classes manually to instantiate. There is just need to supply container environment (e.g. inject this.applicationContext to the listenerContainer object).

For the declaration on the Broker you must use RabbitAdmin bean from your applicationContext.

From other side there is no reason to start a new listenerContainer manually. The existing one can supplied with your new Queue at runtime.