Specifying timeout for reading messages from activemq queue using camel

Question

I am using camel to read messages from an activemq queue, process it and post it to another queue. The route looks as follows:

from("jms:incoming.queue")
.process(new MyProcessor())
.to("jms:outgoing.queue");

I need to specify a timeout such that if there are no messages in "incoming.queue" for more than 3 minutes, I would like to stop the route. I can use OnCompletion() but it gets called after each message. I can specify timeout for sending message to "outgoing.queue". Is there a way I can specify a timeout such that if there are no message for more than 3 minutes in the "incoming.queue", I can stop the route?

Thanks in advance for you help.

Why would you need to stop the route which is not active for more than 3 minutes? You could write last processing time somewhere and check it with other route pushed with timer component. — white
@white - The requirement is that the messages in the "incoming.queue" needs to be processed during off hours, say at 8:00PM in the evening. The route needs to process all the messages in the queue at that time. Once processing all the messages, the route needs to be stopped so that it does not start processing the messages as soon as they starting arriving in the queue next day. The route needs to be started again in the evening at 8:00PM. Is there an easy way to do it? — Sathish

Ben ODay Ben ODay · Accepted Answer · 2012-10-11T21:40:56

two options I can think of...

use a CronScheduledRoutePolicy to start/stop your route automatically at specified times...

CronScheduledRoutePolicy myPolicy = new CronScheduledRoutePolicy();
myPolicy.setRouteStartTime("0 20 * * * ?");
myPolicy.setRouteStopTime("0 0 * * * ?");

from("jms:incoming.queue")
    .routePolicy(myPolicy).noAutoStartup()
    .process(new MyProcessor())
    .to("jms:outgoing.queue");

use a camel-quartz route and a polling consumer to drain the queue on a schedule

MyCoolBean cool = new MyCoolBean();
cool.setProducer(template); 
cool.setConsumer(consumer);

from("quartz://myGroup/myTimerName?cron=0+20+*+*+*+?")
    .bean(cool);

//MyCoolBean snippet 
while (true) {
    // receive the message from the queue, wait at most 60s
    Object msg = consumer.receiveBody("jms:incoming.queue", 60000);
    if (msg == null) {
        break;
    }
    producer.sendBody("jms:outgoing.queue", msg);
}

Specifying timeout for reading messages from activemq queue using camel

2 Answers