Java 类javax.jms.MessageListener 实例源码

@Test
public void on_SUBSCRIBE() throws JMSException {
    final Session session = mock(Session.class);
    ReflectionUtil.set(this.connection, "session", session);
    final Destination destination = mock(Destination.class);
    when(session.toDestination("/dest")).thenReturn(destination);
    when(session.getConnection()).thenReturn(this.connection);
    final MessageConsumer consumer = mock(MessageConsumer.class);
    when(session.createConsumer(eq(destination), any(String.class))).thenReturn(consumer);

    this.connection.on(new Message("ABC123", Frame.subscribe("1", "/dest").build()));

    verify(this.log).info("Message received from client. [sessionId={},command={}]", "ABC123", Command.SUBSCRIBE);
    verify(session).toDestination("/dest");
    verify(session).getConnection();
    verify(session).createConsumer(destination, "session IS NULL OR session = 'ABC123'");
    verify(consumer).setMessageListener(any(MessageListener.class));
    verifyNoMoreInteractions(session, destination, consumer);
}

/**
 * Invoke the specified listener: either as standard JMS MessageListener
 * or (preferably) as Spring SessionAwareMessageListener.
 * @param session the JMS Session to operate on
 * @param message the received JMS Message
 * @throws JMSException if thrown by JMS API methods
 * @see #setMessageListener
 */
@SuppressWarnings("rawtypes")
protected void invokeListener(Session session, Message message) throws JMSException {
    Object listener = getMessageListener();

    if (listener instanceof SessionAwareMessageListener) {
        doInvokeListener((SessionAwareMessageListener) listener, session, message);
    }
    else if (listener instanceof MessageListener) {
        doInvokeListener((MessageListener) listener, message);
    }
    else if (listener != null) {
        throw new IllegalArgumentException(
                "Only MessageListener and SessionAwareMessageListener supported: " + listener);
    }
    else {
        throw new IllegalStateException("No message listener specified - see property 'messageListener'");
    }
}

@Test
public void setupJmsMessageContainerFullConfig() {
    DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
    MessageListener messageListener = new MessageListenerAdapter();
    SimpleJmsListenerEndpoint endpoint = new SimpleJmsListenerEndpoint();
    endpoint.setDestination("myQueue");
    endpoint.setSelector("foo = 'bar'");
    endpoint.setSubscription("mySubscription");
    endpoint.setConcurrency("5-10");
    endpoint.setMessageListener(messageListener);

    endpoint.setupListenerContainer(container);
    assertEquals("myQueue", container.getDestinationName());
    assertEquals("foo = 'bar'", container.getMessageSelector());
    assertEquals("mySubscription", container.getSubscriptionName());
    assertEquals(5, container.getConcurrentConsumers());
    assertEquals(10, container.getMaxConcurrentConsumers());
    assertEquals(messageListener, container.getMessageListener());
}

@Test
public void setupJcaMessageContainerFullConfig() {
    JmsMessageEndpointManager container = new JmsMessageEndpointManager();
    MessageListener messageListener = new MessageListenerAdapter();
    SimpleJmsListenerEndpoint endpoint = new SimpleJmsListenerEndpoint();
    endpoint.setDestination("myQueue");
    endpoint.setSelector("foo = 'bar'");
    endpoint.setSubscription("mySubscription");
    endpoint.setConcurrency("10");
    endpoint.setMessageListener(messageListener);

    endpoint.setupListenerContainer(container);
    JmsActivationSpecConfig config = container.getActivationSpecConfig();
    assertEquals("myQueue", config.getDestinationName());
    assertEquals("foo = 'bar'", config.getMessageSelector());
    assertEquals("mySubscription", config.getSubscriptionName());
    assertEquals(10, config.getMaxConcurrency());
    assertEquals(messageListener, container.getMessageListener());
}

@Test
public void createSimpleContainer() {
    SimpleJmsListenerContainerFactory factory = new SimpleJmsListenerContainerFactory();
    setDefaultJmsConfig(factory);
    SimpleJmsListenerEndpoint endpoint = new SimpleJmsListenerEndpoint();

    MessageListener messageListener = new MessageListenerAdapter();
    endpoint.setMessageListener(messageListener);
    endpoint.setDestination("myQueue");

    SimpleMessageListenerContainer container = factory.createListenerContainer(endpoint);

    assertDefaultJmsConfig(container);
    assertEquals(messageListener, container.getMessageListener());
    assertEquals("myQueue", container.getDestinationName());
}

@Test
public void createJmsContainerFullConfig() {
    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    setDefaultJmsConfig(factory);
    factory.setCacheLevel(DefaultMessageListenerContainer.CACHE_CONSUMER);
    factory.setConcurrency("3-10");
    factory.setMaxMessagesPerTask(5);

    SimpleJmsListenerEndpoint endpoint = new SimpleJmsListenerEndpoint();
    MessageListener messageListener = new MessageListenerAdapter();
    endpoint.setMessageListener(messageListener);
    endpoint.setDestination("myQueue");
    DefaultMessageListenerContainer container = factory.createListenerContainer(endpoint);

    assertDefaultJmsConfig(container);
    assertEquals(DefaultMessageListenerContainer.CACHE_CONSUMER, container.getCacheLevel());
    assertEquals(3, container.getConcurrentConsumers());
    assertEquals(10, container.getMaxConcurrentConsumers());
    assertEquals(5, container.getMaxMessagesPerTask());

    assertEquals(messageListener, container.getMessageListener());
    assertEquals("myQueue", container.getDestinationName());
}

@Test
public void createJcaContainerFullConfig() {
    DefaultJcaListenerContainerFactory factory = new DefaultJcaListenerContainerFactory();
    setDefaultJcaConfig(factory);
    factory.setConcurrency("10");

    SimpleJmsListenerEndpoint endpoint = new SimpleJmsListenerEndpoint();
    MessageListener messageListener = new MessageListenerAdapter();
    endpoint.setMessageListener(messageListener);
    endpoint.setDestination("myQueue");
    JmsMessageEndpointManager container = factory.createListenerContainer(endpoint);

    assertDefaultJcaConfig(container);
    assertEquals(10, container.getActivationSpecConfig().getMaxConcurrency());
    assertEquals(messageListener, container.getMessageListener());
    assertEquals("myQueue", container.getActivationSpecConfig().getDestinationName());
}

protected void createSubscriber(TopicSession topicSession, Topic topic) throws JMSException {
    // Create subscriber
    MessageConsumer subscriber = topicSession.createConsumer(topic);

    // Attach message listener to subscriber
    subscriber.setMessageListener(new MessageListener() {
        public void onMessage(javax.jms.Message message) {
            try {
                // Process the message
                processMessage(message);
            } catch (Exception ex) {
                // Error
                logger.error("Error", ex);
            }
        }
    });
}

public void setupBus(MessageListener msgLis) {
    try {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                ActiveMQConnectionFactory.DEFAULT_BROKER_BIND_URL);
        connection = connectionFactory.createConnection();

        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        destination = session.createTopic("OPCUAMilo");
        //Setup producer
        producer = session.createProducer(destination);

        //setup consumer
        consumer = session.createConsumer(destination);
        connection.start();
        consumer.setMessageListener(msgLis);

    } catch (JMSException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }

}

@Bean
public DefaultMessageListenerContainer messageListener() {
    DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
    container.setConnectionFactory(this.connectionFactory);
    container.setDestinationName("testQueue");
    container.setMessageListener(new MessageListener() {
        @Override
        public void onMessage(Message message) {
            try {
                System.out.println(message.getBody(Object.class));
            }
            catch (JMSException ex) {
                ex.printStackTrace();
            }
        }
    });
    return container;
}

@Bean
public DefaultMessageListenerContainer messageListener() {
    DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
    container.setConnectionFactory(this.connectionFactory);
    container.setDestinationName("testQueue");
    container.setMessageListener(new MessageListener() {
        @Override
        public void onMessage(Message message) {
            try {
                System.out.println(message.getBody(Object.class));
            }
            catch (JMSException ex) {
                ex.printStackTrace();
            }
        }
    });
    return container;
}

public static void receiveMessage(final MessageListener listener, final String brokerHost, final String brokerPort, final String topic) throws JMSException {
    // Create a ConnectionFactory
    final ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(String.format("tcp://%s:%s", brokerHost, brokerPort));

    // Create a Connection
    final Connection connection = connectionFactory.createConnection();
    connection.start();

    // Create a Session
    final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

    // Create the destination (Topic or Queue)
    final Destination destination = session.createQueue(topic);

    // register MessageListener
    final MessageConsumer consumer = session.createConsumer(destination);
    consumer.setMessageListener(listener);
}

public static String receiveMessage(MessageListener listener, String brokerHost, String brokerPort, String topic) throws JMSException {
    // Create a ConnectionFactory
    ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(String.format("tcp://%s:%s", brokerHost, brokerPort));

    // Create a Connection
    Connection connection = connectionFactory.createConnection();
    connection.start();

    // Create a Session
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

    // Create the destination (Topic or Queue)
    Destination destination = session.createQueue(topic);

    // register MessageListener
    MessageConsumer consumer = session.createConsumer(destination);
    Message message = consumer.receive();
    if (message instanceof TextMessage) {
        return ((TextMessage) message).getText();
    } else {
        return null;
    }
}

/**
 * Creates a {@link MessageConsumerResources} with a dedicated
 * {@link Session} required for transacted and InOut consumers.
 */
private MessageConsumerResources createConsumer() throws Exception {
    MessageConsumerResources answer;
    Connection conn = getConnectionResource().borrowConnection();
    try {
        Session session = conn.createSession(isTransacted(), isTransacted() ? Session.SESSION_TRANSACTED : Session.AUTO_ACKNOWLEDGE);
        Destination destination = getEndpoint().getDestinationCreationStrategy().createDestination(session, getDestinationName(), isTopic());
        MessageConsumer messageConsumer = JmsObjectFactory.createMessageConsumer(session, destination, getMessageSelector(), isTopic(), getDurableSubscriptionId());
        MessageListener handler = createMessageHandler(session);
        messageConsumer.setMessageListener(handler);

        answer = new MessageConsumerResources(session, messageConsumer);
    } catch (Exception e) {
        log.error("Unable to create the MessageConsumer", e);
        throw e;
    } finally {
        getConnectionResource().returnConnection(conn);
    }
    return answer;
}

@Bean
public DefaultMessageListenerContainer messageListener() {
    DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
    container.setConnectionFactory(this.connectionFactory);
    container.setDestinationName("testQueue");
    container.setMessageListener(new MessageListener() {
        @Override
        public void onMessage(Message message) {
            try {
                System.out.println(message.getBody(Object.class));
            }
            catch (JMSException ex) {
                ex.printStackTrace();
            }
        }
    });
    return container;
}

/**
 * Set message listener
 *
 * @param listener The listener
 * @throws JMSException Thrown if an error occurs
 */
@Override
public void setMessageListener(final MessageListener listener) throws JMSException {
   session.lock();
   try {
      checkState();
      session.checkStrict();
      if (listener == null) {
         consumer.setMessageListener(null);
      } else {
         consumer.setMessageListener(wrapMessageListener(listener));
      }
   } finally {
      session.unlock();
   }
}

/**
 * Sets the message consumer's <CODE>MessageListener</CODE>.
 * <P>
 * Setting the message listener to null is the equivalent of unsetting the
 * message listener for the message consumer.
 * <P>
 * The effect of calling <CODE>MessageConsumer.setMessageListener</CODE>
 * while messages are being consumed by an existing listener or the consumer
 * is being used to consume messages synchronously is undefined.
 *
 * @param listener the listener to which the messages are to be delivered
 * @throws JMSException if the JMS provider fails to receive the next
 *                 message due to some internal error.
 * @see javax.jms.MessageConsumer#getMessageListener
 */
@Override
public void setMessageListener(MessageListener listener) throws JMSException {
    checkClosed();
    if (info.getPrefetchSize() == 0) {
        throw new JMSException("Illegal prefetch size of zero. This setting is not supported for asynchronous consumers please set a value of at least 1");
    }
    if (listener != null) {
        boolean wasRunning = session.isRunning();
        if (wasRunning) {
            session.stop();
        }

        this.messageListener.set(listener);
        session.redispatch(this, unconsumedMessages);

        if (wasRunning) {
            session.start();
        }
    } else {
        this.messageListener.set(null);
    }
}

/**
 * To receive a message from a queue.
 *
 * @throws JMSException         JMS Exception.
 * @throws InterruptedException Interrupted exception while waiting in between messages.
 */
public void receiveMessagesFromQueue() throws JMSException, InterruptedException {
    QueueConnection queueConn = (QueueConnection) connectionFactory.createConnection();
    QueueSession queueSession = queueConn.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    Destination destination = queueSession.createQueue(JMSTestConstants.QUEUE_NAME_1);
    MessageConsumer queueReceiver = queueSession.createConsumer(destination);
    MessageListener listener = message -> {
        try {
            if (message instanceof TextMessage) {
                TextMessage textMessage = (TextMessage) message;
                logger.info("Message text received : " + (textMessage.getText()));
            }
        } catch (JMSException e) {
            logger.info("JMS exception occurred.");
        }
    };
    queueReceiver.setMessageListener(listener);
    queueConn.start();
}

/**
 * Create A Consumer for the given Queue and set the given
 * {@link MessageListener}.
 *
 * @param destination The destination of the {@link Queue}
 * @param listener The {@link MessageListener} to set
 * @return true means failure
 */
public synchronized boolean setupQueueListener(final String destination, final MessageListener listener) {
    this.checkAllowedState(State.SESSION_ACTIVE);
    final Destination dest = new ActiveMQQueue(destination);
    MessageConsumer consumer;
    try {
        consumer = this.session.createConsumer(dest);
        consumer.setMessageListener(listener);
    }
    catch (final JMSException ex) {
        LOGGER.error("Problems during setup of QueueListener.", ex);
        this.state = State.FAILURE;
        return true;
    }
    return false;
}

/**
 * Create A Consumer for the given Topic and set the given
 * {@link MessageListener}.
 *
 * @param topic The {@link Topic} to use
 * @param listener The {@link MessageListener} to set
 * @return true means failure
 */
public synchronized boolean setupTopicListener(final String topic, final MessageListener listener) {
    this.checkAllowedState(State.SESSION_ACTIVE);
    final ActiveMQTopic top = new ActiveMQTopic(topic);
    MessageConsumer consumer;
    try {
        consumer = this.session.createConsumer(top);
        consumer.setMessageListener(listener);
    }
    catch (final JMSException ex) {
        LOGGER.error("Problems during setup of TopicListener.", ex);
        this.state = State.FAILURE;
        return true;
    }
    return false;
}

public static void main(String[] args) throws Exception {
   ConnectionFactory factory = new ActiveMQJMSConnectionFactory("tcp://localhost:61616");
   Destination destination = ActiveMQDestination.fromPrefixedName("queue://shipping");

   try (Connection conn = factory.createConnection()) {
      Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
      MessageConsumer consumer = session.createConsumer(destination);
      consumer.setMessageListener(new MessageListener() {
         @Override
         public void onMessage(Message message) {
            System.out.println("Received Message: ");
            Order order = Jms.getEntity(message, Order.class);
            System.out.println(order);
         }
      });
      conn.start();
      Thread.sleep(1000000);
   }
}

public static void main(String[] args) throws Exception {
   System.out.println("Receive Setup...");
   ConnectionFactory factory = new ActiveMQJMSConnectionFactory("tcp://localhost:61616");
   Destination destination = ActiveMQDestination.fromPrefixedName("queue://orders");

   try (Connection conn = factory.createConnection()) {
      Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
      MessageConsumer consumer = session.createConsumer(destination);
      consumer.setMessageListener(new MessageListener() {
         @Override
         public void onMessage(Message message) {
            System.out.println("Received Message: ");
            Order order = Jms.getEntity(message, Order.class);
            System.out.println(order);
         }
      });
      conn.start();
      Thread.sleep(1000000);
   }
}

@Test
public void testSendManyMessages() throws Exception {
   conn.connect(defUser, defPass);

   MessageConsumer consumer = session.createConsumer(queue);

   int count = 1000;
   final CountDownLatch latch = new CountDownLatch(count);
   consumer.setMessageListener(new MessageListener() {

      @Override
      public void onMessage(Message arg0) {
         latch.countDown();
      }
   });

   for (int i = 1; i <= count; i++) {
      send(conn, getQueuePrefix() + getQueueName(), null, "Hello World!");
   }

   assertTrue(latch.await(60, TimeUnit.SECONDS));
}

@Test
public void testSendManyMessages() throws Exception {
   MessageConsumer consumer = session.createConsumer(queue);

   conn.connect(defUser, defPass);

   int count = 1000;
   final CountDownLatch latch = new CountDownLatch(count);
   consumer.setMessageListener(new MessageListener() {
      @Override
      public void onMessage(Message arg0) {
         latch.countDown();
      }
   });

   for (int i = 1; i <= count; i++) {
      send(conn, getQueuePrefix() + getQueueName(), null, "Hello World");
   }

   assertTrue(latch.await(60, TimeUnit.SECONDS));

   conn.disconnect();
}

@Test
public void testCantReceiveWhenListenerIsSet() throws Exception {
   conn = cf.createConnection();
   Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
   jBossQueue = ActiveMQJMSClient.createQueue(JmsConsumerTest.Q_NAME);
   MessageConsumer consumer = session.createConsumer(jBossQueue);
   consumer.setMessageListener(new MessageListener() {
      @Override
      public void onMessage(final Message msg) {
      }
   });

   try {
      consumer.receiveNoWait();
      Assert.fail("Should throw exception");
   } catch (JMSException e) {
      // Ok
   }
}

protected List<CompletableFuture<List<Message>>> receiveMessagesAsync(int count, MessageConsumer... consumer) throws JMSException {
   AtomicInteger totalCount = new AtomicInteger(count);
   List<CompletableFuture<List<Message>>> resultsList = new ArrayList<>();
   List<List<Message>> receivedResList = new ArrayList<>();

   for (int i = 0; i < consumer.length; i++) {
      final int index = i;
      resultsList.add(new CompletableFuture<>());
      receivedResList.add(new ArrayList<>());
      MessageListener myListener = message -> {
         System.out.println("Mesages received" + message + " count: " + totalCount.get());
         receivedResList.get(index).add(message);
         if (totalCount.decrementAndGet() == 0) {
            for (int j = 0; j < consumer.length; j++) {
               resultsList.get(j).complete(receivedResList.get(j));
            }
         }
      };
      consumer[i].setMessageListener(myListener);
   }
   return resultsList;
}

public ForwardingClient(int from, int to) throws JMSException {
   toConnection = createConnection(from);
   Session toSession = toConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
   final MessageProducer producer = toSession.createProducer(new ActiveMQQueue("Q" + to));
   producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
   producer.setDisableMessageID(true);

   fromConnection = createConnection(from);
   Session fromSession = fromConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
   MessageConsumer consumer = fromSession.createConsumer(new ActiveMQQueue("Q" + from));

   consumer.setMessageListener(new MessageListener() {
      @Override
      public void onMessage(Message msg) {
         try {
            producer.send(msg);
            forwardCounter.incrementAndGet();
         } catch (JMSException e) {
            // this is caused by the connection getting closed.
         }
      }
   });
}

public void testNoSlowConsumerAdvisory() throws Exception {
   Session s = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
   Queue queue = s.createQueue(getClass().getName());
   MessageConsumer consumer = s.createConsumer(queue);
   consumer.setMessageListener(new MessageListener() {
      @Override
      public void onMessage(Message message) {
      }
   });
   Topic advisoryTopic = AdvisorySupport.getSlowConsumerAdvisoryTopic((ActiveMQDestination) queue);
   s = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
   MessageConsumer advisoryConsumer = s.createConsumer(advisoryTopic);
   // start throwing messages at the consumer
   MessageProducer producer = s.createProducer(queue);
   for (int i = 0; i < MESSAGE_COUNT; i++) {
      BytesMessage m = s.createBytesMessage();
      m.writeBytes(new byte[1024]);
      producer.send(m);
   }
   Message msg = advisoryConsumer.receive(1000);
   assertNull(msg);
}

private void createConsumer() throws JMSException {
   if (durableSub) {
      consumer = session.createDurableSubscriber(topic, durableID);
   } else {
      consumer = session.createConsumer(topic);
   }
   consumer.setMessageListener(new MessageListener() {

      @Override
      public void onMessage(Message arg0) {
         TextMessage msg = (TextMessage) arg0;
         try {
            LOG.debug("Received message [" + msg.getText() + "]");
            receivedStrings.add(msg.getText());
            receivedLatch.countDown();
         } catch (JMSException e) {
            fail("Unexpected :" + e);
         }
      }

   });
}

@Override
public MessageListener getMessageListener() {
    try {
        return consumer.getMessageListener();
    } catch (JMSException e) {
        throw JMSExceptionSupport.createRuntimeException(e);
    }
}

@Override
public void setMessageListener(MessageListener listener) {
    try {
        consumer.setMessageListener(listener);
    } catch (JMSException e) {
        throw JMSExceptionSupport.createRuntimeException(e);
    }
}

@Override
public MessageListener getMessageListener() {
    try {
        return consumer.getMessageListener();
    } catch (JMSException e) {
        throw JMSExceptionSupport.createRuntimeException(e);
    }
}

@Override
public void setMessageListener(MessageListener listener) {
    try {
        consumer.setMessageListener(listener);
    } catch (JMSException e) {
        throw JMSExceptionSupport.createRuntimeException(e);
    }
}

@Test
public void start_agent_listener() throws JMSException {
    client.startAgentListener(TEST_AGENT, agentClientListener);

    verify(jmsClient).addTopicListener(eq(Constants.AGENTS_TOPIC), any(MessageListener.class));
    verify(jmsClient).startListening();
    verifyNoMoreInteractions(jmsClient);
}

@Test
public void startMetricsListener_starts_jms_listener() throws JMSException {
    clusterClient.startMetricsListener(metricsListener);

    verify(jmsClient).addTopicListener(eq(METRICS_TOPIC), any(MessageListener.class));
    verify(jmsClient).startListening();
    verifyNoMoreInteractions(jmsClient);
    verifyZeroInteractions(metricsListener);
}

@Test
public void start_actor_listener() throws JMSException {
    when(actor.getKey()).thenReturn(DUMMY_ACTOR);

    client.startActorListener(actor);

    verify(jmsClient).addQueueListener(eq(DUMMY_ACTOR.getName()), any(MessageListener.class));
    verify(jmsClient).startListening();
    verifyNoMoreInteractions(jmsClient);
}

public <X> void beans(
        final @Observes ProcessAnnotatedType<X> processBean
) {
    final MessageDriven annotation = processBean.getAnnotatedType().getAnnotation(MessageDriven.class);
    if (annotation != null) {
        final Class<X> javaClass = processBean.getAnnotatedType().getJavaClass();
        if (!MessageListener.class.isAssignableFrom(javaClass)) {
            throw new TestEEfiException("The @MessageDriven bean " + javaClass.getName()
                    + " does not implement MessageListener"
            );
        }
        registry.register(annotation, (Class<? extends MessageListener>) javaClass);
    }
}

@Override
public void sendMessage(final String destination, final Message message) {
    final Class<? extends MessageListener> mdbClass = messageDrivenBeanRegistry.get(destination);
    final Releaser releaser = new Releaser();
    final MessageListener mdb = dependencyInjection.getInstanceOf(mdbClass, releaser);
    try {
        mdb.onMessage(message);
    } finally {
        releaser.release();
    }
}

@Test
public void receive() throws Exception {
    when(delegate.getMessageListener()).thenReturn(mock(PreservesMessagePropertiesMessageListener.class));
    propagator.receive();
    verify(delegate).receive();
    reset(delegate);
    when(delegate.getMessageListener()).thenReturn(mock(MessageListener.class));
    propagator.receive();
    verify(delegate).receive();
}

@Test
public void receive1() throws Exception {
    when(delegate.getMessageListener()).thenReturn(mock(PreservesMessagePropertiesMessageListener.class));
    propagator.receive(0);
    verify(delegate).receive(0);
    reset(delegate);
    when(delegate.getMessageListener()).thenReturn(mock(MessageListener.class));
    propagator.receive(0);
    verify(delegate).receive(0);
}