javax.jms.JMSConsumer#setMessageListener ( )源码实例Demo

@After
public void stop() throws Exception {
    JMSContext context = factory.createContext();
    Destination channel = context.createQueue(MockEvent.class.getName());
    JMSConsumer consumer = context.createConsumer(channel);
    // 清理测试消息
    logger.info("清理JMS测试消息开始");
    AtomicInteger count = new AtomicInteger();
    consumer.setMessageListener((data) -> {
        String message = StringUtility.format("清理JMS测试消息[{}]", count.incrementAndGet());
        logger.info(message);
    });
    Thread.sleep(1000L);
    logger.info("清理JMS测试消息结束");
    factory.close();
}
 

@Override
public CompletableFuture<Message> registerConcreteListenerImplementation(Destination destination) {
    JMSConsumer consumer = context.createConsumer(destination);
    final CompletableFuture<Message> incomingMessageFuture = new CompletableFuture<>();
    //noinspection Convert2Lambda,Anonymous2MethodRef
    consumer.setMessageListener(new MessageListener() {
        @Override
        public void onMessage(Message message) {
            incomingMessageFuture.complete(message);
        }
    });
    return incomingMessageFuture;
}
 

@Override
public CompletableFuture<Message> registerListenerLambda(Destination destination) {
    JMSConsumer consumer = context.createConsumer(destination);
    final CompletableFuture<Message> incomingMessageFuture = new CompletableFuture<>();
    // ActiveMQ Artemis wraps listeners with actual MessageListener instances
    // of org.apache.activemq.artemis.jms.client.ActiveMQJMSConsumer$MessageListenerWrapper anyway..
    //noinspection Convert2MethodRef
    consumer.setMessageListener(message -> incomingMessageFuture.complete(message));
    return incomingMessageFuture;
}
 

@Override
public CompletableFuture<Message> registerListenerMethodReference(Destination destination) {
    JMSConsumer consumer = context.createConsumer(destination);
    final CompletableFuture<Message> incomingMessageFuture = new CompletableFuture<>();
    // ActiveMQ Artemis wraps listeners with actual MessageListener instances
    // of org.apache.activemq.artemis.jms.client.ActiveMQJMSConsumer$MessageListenerWrapper anyway..
    consumer.setMessageListener(incomingMessageFuture::complete);
    return incomingMessageFuture;
}
 

MyJmsClient(Destination destination) {
    JMSConsumer consumer = jms.createConsumer(destination);
    consumer.setMessageListener(messages::add);
}
 

MyJmsClient(Destination destination) {
    JMSConsumer consumer = jms.createConsumer(destination);
    consumer.setMessageListener(messages::add);
}
 

@Test
public void testContextStopAndCloseFromMessageListeners() throws Exception {
   final JMSContext context1 = context.createContext(Session.AUTO_ACKNOWLEDGE);
   JMSConsumer consumer1 = context1.createConsumer(queue1);

   final CountDownLatch latch1 = new CountDownLatch(1);

   InvalidMessageListener listener1 = new InvalidMessageListener(context1, latch1, 1);

   consumer1.setMessageListener(listener1);

   JMSProducer producer = context1.createProducer();
   Message msg = context1.createTextMessage("first message");
   producer.send(queue1, msg);

   latch1.await();

   Throwable error1 = listener1.getError();

   assertNotNull(error1);

   assertTrue(error1 instanceof IllegalStateRuntimeException);

   context1.close();

   final JMSContext context2 = context.createContext(Session.AUTO_ACKNOWLEDGE);
   JMSConsumer consumer2 = context2.createConsumer(queue1);

   final CountDownLatch latch2 = new CountDownLatch(1);

   InvalidMessageListener listener2 = new InvalidMessageListener(context2, latch2, 2);

   consumer2.setMessageListener(listener2);

   JMSProducer producer2 = context2.createProducer();
   Message msg2 = context2.createTextMessage("second message");
   producer2.send(queue1, msg2);

   latch2.await();

   Throwable error2 = listener2.getError();

   assertNotNull(error2);

   assertTrue(error2 instanceof IllegalStateRuntimeException);

   context2.close();
}