programmatic RabbitTemplate configuration

[Gary Russell]

Have you looked at using HA/mirrored queues? Rabbit takes care of the mirroring and queue declaration for you. One of the drivers (aside from the reduction in the use of temporary queues) for implementing a fixed reply-queue was to enable it to be used with HA arguments.

[cogitate]

Thanks Gary!
will look at HA/mirrored queues - however, at this point it might be too much(apologize in advance for non-quantitative adjective) to implement from an operations perspective.

i was hoping for maybe a listener in rabbitadmin that i can use to re-init the queues when the listener re-inits.
if that's at all possible.
regards,
-monish

[Gary Russell]

How are you declaring your queues? The RabbitAdmin (in afterPropertiesSet) sets itself up as a ConnectionListener and, when a new connection is established (after failover), it will re-declare the queues.

[cogitate]

// this is the client that's pre-configured and pooled
import..
class SpringClient
{
...
...
private Queue replyQueue = null;

public SpringClient(CachingConnectionFactory connectionFactory, RabbitAdmin admin)
{
this.connectionFactory = connectionFactory;
this.admin = admin;
amqpTemplate = new RabbitTemplate(connectionFactory);
replyContainer = new SimpleMessageListenerContainer(connectionFactory);
replyQueueName = "REPLY_QUEUE."+ UUID.randomUUID().toString();
}

public Message callService(String exchange,String serviceName,Message payload) throws Exception
{
// create reply queue just in time
if (replyQueue == null)
{
try
{
createReplyQueue();
}
catch(Exception e) {}
}
amqpTemplate.setExchange(exchange);
amqpTemplate.setRoutingKey(serviceName);
amqpTemplate.setReplyQueue(replyQueue);
Message replyMessage = null;
try
{
replyMessage = amqpTemplate.sendAndReceive(payload);
}
catch(Exception e) { .. }
....
return replyMessage;
}

private void createReplyQueue() throws IOException
{
declareQueue();
replyContainer.setQueueNames(replyQueueName);
replyContainer.setMessageListener(amqpTemplate);
replyContainer.afterPropertiesSet();
replyContainer.start();
}

private void declareQueue() throws IOException
{
admin = new RabbitAdmin(connectionFactory);
boolean durable = false;
boolean autoDelete = true;
boolean exclusive = true;
replyQueue = new Queue(replyQueueName,durable,exclusive,autoDelete) ;
admin.declareQueue(replyQueue);
}
}

so if i could check
if (replyQueue == null && replyContainer.reset() )// or something equivalent then i can redeclare the queue

regards,
-monish

[Gary Russell]

Please use [ code ] ... [ /code ] tags around your code/config (no spaces inside brackets); it makes it easier to read.

If you simply let Spring take care of everything, it will just work automatically - the RabbitAdmin is a ConnectionListener and will declare everything for you when the connection is established. For example, from the Spring Integration amqp sample....

    <rabbit:connection-factory id="connectionFactory" />

    <rabbit:template id="amqpTemplate" connection-factory="connectionFactory" />

    <rabbit:admin connection-factory="connectionFactory" />

    <rabbit:queue name="si.test.queue" />

    <rabbit:direct-exchange name="si.test.exchange">
        <rabbit:bindings>
            <rabbit:binding queue="si.test.queue" key="si.test.binding" />
        </rabbit:bindings>
    </rabbit:direct-exchange>

Whenever a connection is established, all the declarations are performed.

[cogitate]

hi gary:
sorry about not putting code in code quotes.
the problem is this "stand-alone" library requires "pre-created" queues for a web-app. that's the whole reason to go into "programmatic" configuration.
will you advice what i should look up so i can become a subscriber to the lifecycle monitor listener?

thanks for your patience and help,
-monish

[Gary Russell]

No problem - I suggest you take a look at the code in RabbitAdmin.afterPropertiesSet() - you'll see how he registers a class as a ConnectionListener, and how that is invoked to run the declarations whenever a new Connection is established.

[cogitate]

thanks gary for the pointers!
i took a shortcut, in the init method i did this:

private void init()
    {
        isUsable = true;
        amqpTemplate = new RabbitTemplate(connectionFactory);
        replyContainer = new SimpleMessageListenerContainer(connectionFactory);
        context = new GenericApplicationContext();
        admin = new RabbitAdmin(connectionFactory);
        admin.setApplicationContext(context);
        admin.setAutoStartup(true);
        replyQueueName = "REPLY_QUEUE."+ UUID.randomUUID().toString();
        createReplyQueue();
    }

..and in createReplyQueue()

 private void createReplyQueue()
    {
        declareQueue();
        replyContainer.setQueueNames(replyQueueName);
        replyContainer.setMessageListener(amqpTemplate);
        replyContainer.afterPropertiesSet();
        replyContainer.start();
    }

    private void declareQueue()    
    {
        boolean durable = false;
        boolean autoDelete = true;
        boolean exclusive = true;
        replyQueue = new Queue(replyQueueName,durable,exclusive,autoDelete);
        context.getBeanFactory().registerSingleton(replyQueueName, replyQueue);
        admin.afterPropertiesSet();

    }

this works very well and passes all the tests. any potential overhead might be only due to creating GenericApplicationContext at initialization time. not sure if that's too much.
RabbitAdmin is a constructor parameter and the manager now hands out pre-configured with a connectionFactory.
The user is allowed to specify #of connections (for performance).

when you get a chance, feel free to point as many errors/bugs/in-efficiencies with this code!
best regards and thanks much,
-cogitate