CODEWITHSUNDEEP
javaspringjmsapache-camel
kdye43
kdye43

Reputation: 25

Camel JMS Request Reply issue with Remote MessageListener

I'm testing out examples of using the JMS Request/Reply with Camel and ActiveMQ. I can get the example to work when camel creates the listener for you. ie.

from("direct:entryPoint").inOut("jms:queue:A");

from("jms:queue:A").
    process(new Processor() {

        @Override
        public void process(Exchange exchange) throws Exception {
            exchange.getIn().setBody("Hello World.");

        }
    });



The issue I'm having now is that I can't get the JMS Request/Reply working with a MessageListener that exists outside of Camel's jvm. The connection times out waiting for a reply. I made sure that the MessageListener is sending the reply to the replyTo queue and I'm also setting the correlationId. What am I doing wrong here? I've googled for days trying to figure this out with no luck. Thanks ahead of time for your help.

Below is the route I'm using and I also put the MessageListener logic below as well.

from("direct:entryPoint").
  inOut("jms:queue:B?concurrentConsumers=4&requestTimeout=240000");

MessageListener onMessage for queue B:

@Override
public void onMessage(Message message) {

    String msg = null;
    ObjectMapper mapper = new ObjectMapper();
    mapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, true);
    String jsonOutput = null;

    try{
        msg = ((TextMessage) message).getText();

        //convert message payload to purchase order
        PurchaseOrder order = mapper.readValue(msg, PurchaseOrder.class);

        //Set the id to see if the request reply worked.
        order.setOrderId(BigInteger.valueOf(111111111));


        if(message.getJMSReplyTo() != null){
            Map<String, Object> headers = new HashMap<String, Object>();
            headers.put("JMSCorrelationID", message.getJMSCorrelationID());
            headers.put("JMSReplyTo", message.getJMSReplyTo().toString());  
            jsonOutput = mapper.writeValueAsString(order);

            //Camel runs on the external jvm so leverage the producerTemplate.
            producerTemplate.
              sendBodyAndHeaders("jms:"+ message.getJMSReplyTo().toString(), 
              jsonOutput, headers);
        }
    }
    catch(Exception e){
        logger.fatal(e.getMessage());

        try {
            if(message.getJMSReplyTo() != null){
                Map<String, Object> headers = new HashMap<String, Object>();
                headers.put("JMSCorrelationID", message.getJMSCorrelationID());

                producerTemplate.
                 sendBodyAndHeaders("jms:"+ message.getJMSReplyTo().toString(),
 e.getMessage(), headers);
            }
        } catch (JMSException e1) {
            logger.fatal(e1.getMessage());
        } catch (Exception e1) {
            logger.fatal(e1.getMessage());
        }
    }
}

Upvotes: 0

Views: 2453

Answers (1)

Petter Nordlander
Petter Nordlander

Reputation: 22279

sendBodyAndHeaders("jms:"+ message.getJMSReplyTo().toString(), 
          jsonOutput, headers);

Would typically resolve to jms:queue://someQueue which might mess things up. Using a javax.jms.Destination converted to string is typically not a good idea unless you take precausions.

You can use the Camel header CamelJmsDestination: headers.put("CamelJmsDestination",message.getJMSReplyTo());

If that works or not, I don't know. Genereally, try to use ActiveMQ with the Web console ( or connect to ActiveMQ with JMX (jconsole)) to look at the queues and try to figure out who is reading what queue and where messages ends up. It's really helpful.

Upvotes: 1

Related Questions