title: Consuming message from ActiveMQ and Publishing to Azure Service Bus Topic
body: <p>I have a requirement to consume JMS message from ActiveMQ queue and publish it to Azure Service Bus Topic. I have used Spring Boot for this project and using Apache qpid to connect Service Bus over AMQP protocol. </p>
  
  <p>The problem I am facing is, after consuming message from ActiveMQ during publishing message to Service Bus Topic if for any reason exception occur I am not able to retain message in ActiveMQ. I want ActiveMQ to hold message until I am able to publish message in Service Bus. I tried using acknowledgement as Session.CLIENT_ACKNOWLEDGE in listener container factory but somewhere from exception it is returning as success which cause ActiveMQ to drop message. However ActiveMQ is able to retry sending 7 time but after that it just drop message. </p>
  
  <p>Below is my code detail.
  I to have to use 2 configuration class to connect ActiveMQ and Service Bus which are as below.</p>
  
  <pre><code>@Configuration
  @EnableJms
  public class RecevierConfig {
  
      private final Logger logger = LoggerFactory.getLogger(RecevierConfig.class);
  
      @Value("${activemq.host}")
      private String activemqHost;
  
      @Value("${activemq.protocol}")
      private String activemqProtocol;
  
      @Value("${activemq.port}")
      private String activemqPort;
  
        @Bean
        public ActiveMQConnectionFactory activeMQConnectionFactory() {
          ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory();
          logger.info("Broker URL : "+activemqProtocol+"://"+activemqHost+":"+activemqPort);
          activeMQConnectionFactory.setBrokerURL(activemqProtocol+"://"+activemqHost+":"+activemqPort);
  
          return activeMQConnectionFactory;
        }
  
        @Bean
        public DefaultJmsListenerContainerFactory jmsListenerContainerFactory() {
          DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
          factory.setConnectionFactory(activeMQConnectionFactory());
          factory.setConcurrency("3-10");
          factory.setSessionAcknowledgeMode(Session.CLIENT_ACKNOWLEDGE);
          factory.setErrorHandler(new DefaultErrorHandler());
          return factory;
  
        }
  
        @Bean
        public QueueReceiver receiver() {
          return new QueueReceiver();
        }
  
  }
  </code></pre>
  
  <p>Below is to create ConnectionFactory and jmsTemplate for Service Bus connection to publish message. </p>
  
  <pre><code>@Configuration
  public class MessagingConfig {
  
        @Value("${spring.application.name}")
          private String clientId;
  
  
          @Bean
          @Primary
          public ConnectionFactory jmsConnectionFactory(MessageStoreDetails details) throws UnsupportedEncodingException {
              JmsConnectionFactory jmsConnectionFactory = new JmsConnectionFactory(details.getSBUrlString());
              jmsConnectionFactory.setUsername(details.getUsername());
              jmsConnectionFactory.setPassword(details.getPassword());
              jmsConnectionFactory.setClientID(clientId);
             // jmsConnectionFactory.setReceiveLocalOnly(true);
              return new CachingConnectionFactory(jmsConnectionFactory);
          }
  
          @Bean
          public JmsTemplate jmsTemplate(ConnectionFactory jmsConnectionFactory) {
              JmsTemplate returnValue = new JmsTemplate();
              returnValue.setConnectionFactory(jmsConnectionFactory);
              return returnValue;
          }
  
  }
  </code></pre>
  
  <p>Below is the code for QueueReceiver receiving message from ActiveMQ</p>
  
  <pre><code>@Service
  public class QueueReceiver {
  
      private final Logger logger = LoggerFactory.getLogger(QueueReceiver.class);
  
      @Autowired
      private TopicPublisher topicPublisher;
  
      @JmsListener(destination = "${activemq.queue}")
      public void receiveMessage(Message message) {
  
          try {
              logger.info("*****Received message from ActiveMQ queue : " + message.getJMSMessageID());
              topicPublisher.sendMessage(message);
              message.acknowledge();
  
          } catch (Exception e) {
              //logger.error(e.toString());
              throw new MessageProcessingException(e.toString());
          }
      }
  }
  </code></pre>
  
  <p>Finally below is the class to publish message </p>
  
  <pre><code>@Service
  public class TopicPublisher {
  
      private final Logger logger = LoggerFactory.getLogger(TopicPublisher.class);
  
      @Value("${servicebus.topic}")
      private String topic;
  
      @Autowired
      private JmsTemplate jmsTemplate;
  
      public void sendMessage(javax.jms.Message message) throws Exception {
             logger.info("#### Sending message to Service Bus Topic : "+ topic +" Message ID: "+ message.getJMSMessageID());
             jmsTemplate.convertAndSend(topic, message);
             logger.info("**** Message Sent to Service Bus Topic : "+ topic +" Message ID: "+ message.getJMSMessageID());
      }
  
  }
  </code></pre>