title: Active MQ + Spring 3.0 - Request/Response Implementation - Sample Program got response But Message Consumer count keps going up and waiting for ever
body: <p>I am trying sample projects to implement a Spring 3 + Active MQ Request/Response Synchronos.. I created a spring configuration file, a message producer that puts message in a queue and message consumer that consumes the message and returns a response...</p>
  
  <p>I am getting the response back... But my sample program doesnt seem to end... When i check the Apache Active MQ Admin console I see that the NUmber of Consumers count keeps going up every time I run my test class... I got to terminate it manually in eclipse for the count to go down in the admin console... </p>
  
  <p>I referred this thread here - <a href="https://stackoverflow.com/questions/8006891/javax-jms-messageconsumer-hangs-on-receive-when-consumer-closed">Stack Overflow Thread</a> - Somebody who has faced the same issue.. But looks like I already have that solution in place and still don't see my issue solved</p>
  
  <p>Also I referred <a href="http://codedependents.com/2010/03/04/synchronous-request-response-with-activemq-and-spring/" rel="nofollow noreferrer">here</a> and <a href="http://codedependents.com/2009/10/16/efficient-lightweight-jms-with-spring-and-activemq/" rel="nofollow noreferrer">here</a> to create my solution</p>
  
  <p>So here is my code</p>
  
  <pre><code>    &lt;!-- creates an activemq connection factory using the amq namespace --&gt;
  
  &lt;bean id="jmsConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"&gt;
      &lt;property name="brokerURL" value="tcp://localhost:61616" /&gt;
  &lt;/bean&gt;
  
  &lt;!-- 
  &lt;bean id="connectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory"
      init-method="start" destroy-method="stop"&gt;
      &lt;property name="connectionFactory" ref="jmsConnectionFactory" /&gt;
      &lt;property name="maxConnections" value="100" /&gt;
  &lt;/bean&gt;
   --&gt;
  
  &lt;!-- CachingConnectionFactory Definition, sessionCacheSize property is the 
      number of sessions to cache --&gt;
  
  &lt;bean id="connectionFactory"
      class="org.springframework.jms.connection.CachingConnectionFactory"&gt;
      &lt;constructor-arg ref="jmsConnectionFactory" /&gt;
      &lt;property name="exceptionListener" ref="jmsExceptionListener" /&gt;
      &lt;property name="sessionCacheSize" value="1" /&gt;
      &lt;property name="cacheConsumers" value="false" /&gt;
      &lt;property name="cacheProducers" value="false" /&gt;
  &lt;/bean&gt;
  
  &lt;!-- JmsTemplate Definition --&gt;
  &lt;bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"&gt;
      &lt;constructor-arg ref="connectionFactory" /&gt;
  
  &lt;/bean&gt;
  
  
  &lt;jms:listener-container connection-factory="connectionFactory"&gt;
      &lt;jms:listener id="request.queue.listener" destination="test.request"
          ref="testMessageListener" /&gt;
  &lt;/jms:listener-container&gt;
  
  &lt;bean id="WorkerClient" class="com.vzwcorp.legal.eplm.active.mq.framework.WorkerClient" /&gt;
  </code></pre>
  
  <p>Requestor Class</p>
  
  <pre><code>@Component
  </code></pre>
  
  <p>public class Requestor {</p>
  
  <pre><code>private static final class ProducerConsumer implements
          SessionCallback&lt;Message&gt; {
  
      private static final int TIMEOUT = 5000;
  
      private final String msg;
  
      private final DestinationResolver destinationResolver;
  
      private final String queue;
  
      public ProducerConsumer(final String msg, String queue,
              final DestinationResolver destinationResolver) {
          this.msg = msg;
          this.queue = queue;
          this.destinationResolver = destinationResolver;
      }
  
      public Message doInJms(final Session session) throws JMSException {
          MessageConsumer consumer = null;
          MessageProducer producer = null;
          try {
              final String correlationId = UUID.randomUUID().toString();
              final Destination requestQueue = destinationResolver
                      .resolveDestinationName(session, queue + ".request",
                              false);
              final Destination replyQueue = destinationResolver
                      .resolveDestinationName(session, queue + ".response",
                              false);
              // Create the consumer first!
              consumer = session.createConsumer(replyQueue,
                      "JMSCorrelationID = '" + correlationId + "'");
              final TextMessage textMessage = session.createTextMessage(msg);
              textMessage.setJMSCorrelationID(correlationId);
              textMessage.setJMSReplyTo(replyQueue);
              // Send the request second!
              producer = session.createProducer(requestQueue);
              producer.send(requestQueue, textMessage);
              // Block on receiving the response with a timeout
              return consumer.receive(TIMEOUT);
          } finally {
              // Don't forget to close your resources
              JmsUtils.closeMessageConsumer(consumer);
              JmsUtils.closeMessageProducer(producer);
          }
      }
  }
  
  private final JmsTemplate jmsTemplate;
  
  @Autowired
  public Requestor(final JmsTemplate jmsTemplate) {
      this.jmsTemplate = jmsTemplate;
  }
  
  public String request(final String request, String queue) {
      // Must pass true as the second param to start the connection
      TextMessage message = (TextMessage) jmsTemplate.execute(
              new ProducerConsumer(request, queue, jmsTemplate
                      .getDestinationResolver()), true);
      try {
          return message.getText();
      } catch (JMSException e) {
          // TODO Auto-generated catch block
          e.printStackTrace();
          return "exception in requestor";
      }
  }
  </code></pre>
  
  <p>}</p>
  
  <p>Message Listener Class</p>
  
  <pre><code>@Component
  </code></pre>
  
  <p>public class TestMessageListener implements MessageListener {</p>
  
  <pre><code>@Autowired
  private WorkerClient WorkerClient;
  
  @Override
  public void onMessage(Message arg0) {
      WorkerClient.delegateToClient(arg0);
  
  }
  </code></pre>
  
  <p>}</p>
  
  <p>Worker Client Class</p>
  
  <pre><code>@Component
  </code></pre>
  
  <p>public class WorkerClient implements ApplicationContextAware {</p>
  
  <pre><code>private ApplicationContext ctx;
  
  private JmsTemplate jmsTemplate;
  
  public void delegateToClient(Message arg0) {
      MessageProducer producer = null;
      if (arg0 instanceof TextMessage) {
          try {
              final TextMessage message = (TextMessage) arg0;
  
              System.out.println("Message received by Listener: "
                      + message.getJMSCorrelationID() + " - "
                      + message.getText());
  
              jmsTemplate.setDefaultDestination(message.getJMSReplyTo());
  
              Session session = jmsTemplate.getConnectionFactory()
                      .createConnection()
                      .createSession(false, Session.AUTO_ACKNOWLEDGE);
              producer = session.createProducer(message.getJMSReplyTo());
              final TextMessage textMessage = session
                      .createTextMessage("I did it at last");
              textMessage.setJMSCorrelationID(message.getJMSCorrelationID());
              textMessage.setJMSReplyTo(message.getJMSReplyTo());
              producer.send(message.getJMSReplyTo(), textMessage);
  
          } catch (Exception e) {
              e.printStackTrace();
  
          } finally {
              JmsUtils.closeMessageProducer(producer);
          }
      }
  }
  
  @Override
  public void setApplicationContext(ApplicationContext arg0)
          throws BeansException {
      ctx = arg0;
      this.jmsTemplate = (JmsTemplate) ctx.getBean("jmsTemplate");
  }
  </code></pre>
  
  <p>At Last The test class</p>
  
  <pre><code>public class TestSync {
  
  public static void main(String[] args) {
  
      ApplicationContext ctx = new ClassPathXmlApplicationContext(
              "activeMQConfiguration.xml");
      OrderService orderService = (OrderService) ctx.getBean("orderService");
      JmsTemplate jmsTemplate = (JmsTemplate) ctx.getBean("jmsTemplate");
       Requestor req = new Requestor(jmsTemplate);
      //CopyOfRequestor req = new CopyOfRequestor(jmsTemplate);
      String response = req.request("Hello World", "test");
      System.out.println(response);
  
  }
  </code></pre>
  
  <p>}</p>
  
  <p>So how to fix the message consumer returning back and my test class ending? Please Help....</p>