测试JMS和Spring集成

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了测试JMS和Spring集成相关的知识,希望对你有一定的参考价值。

我正在尝试写下测试类,以便测试在JMS队列上侦听的消息驱动通道适配器是否将消息转发到正确的通道(参考Advanced Spring Integration Testing)。以下是测试上下文xml:

<!--  MockRunner configuration  -->
    <bean id="destinationManager" class="com.mockrunner.jms.DestinationManager"/>

    <bean id="outgoingDestination" factory-bean="destinationManager" factory-method="createQueue">
        <constructor-arg index="0" value="demoMockRunnerQueue"/>
    </bean>

    <bean id="configurationManager" class="com.mockrunner.jms.ConfigurationManager"/>

    <bean id="connectionFactory" class="com.mockrunner.mock.jms.MockQueueConnectionFactory">
        <constructor-arg index="0" ref="destinationManager"/>
        <constructor-arg index="1" ref="configurationManager"/>
    </bean>

    <!--  Spring JMS Template -->
    <bean id="jmsTemplate" class="org.mockito.Mockito" factory-method="mock">
        <constructor-arg value="org.springframework.jms.core.JmsTemplate" />
    </bean>

以下是带有消息驱动程序通道的spring集成配置:

<int:channel id="inbound"/>

<int-jms:message-driven-channel-adapter id="jmsIn"
                                            channel="inbound"
                                            destination="outgoingDestination"
                                            connection-factory="connectionFactory"
                                            acknowledge="transacted"/>

<int:service-activator input-channel="inbound"
                            ref="messageQueueConsumer"
                            method="consumeMessage"/>

<bean id="messageQueueConsumer" class="uk.co.example.consumer.SimpleMessageConsumer">
    </bean>

接下来是包含测试的java类:

@Resource
JmsTemplate jmsTemplate;

/**
 * "inbound" is the channel used to trigger the service activator (i.e. the message consumer)
 * */
@Resource
@Qualifier("inbound")
SubscribableChannel inbound;

private static final Logger LOGGER = Logger.getLogger(InboundChannelFlowUnitTest.class);

/**
 * This test verifies that a message received on a polling JMS inbound channel adapter is
 * routed to the designated channel and that the message payload is as expected
 *
 * @throws JMSException
 * @throws InterruptedException
 * @throws IOException
 */
@Test
public void testReceiveMessage() throws JMSException, InterruptedException, IOException {
    String msg = "hello";

    boolean sent = verifyJmsMessageReceivedOnChannel(msg, inbound, new CountDownHandler() {

                @Override
                protected void verifyMessage(Message<?> message) {
                    assertEquals("hello", message.getPayload());
                }
            }
    );
    assertTrue("message not sent to expected output channel", sent);
}

/**
 * Provide a message via a mock JMS template and wait for the default timeout to receive the message on the expected channel
 * @param obj The message provided to the poller (currently must be a String)
 * @param expectedOutputChannel The expected output channel
 * @param handler An instance of CountDownHandler to handle (verify) the output message
 * @return true if the message was received on the expected channel
 * @throws JMSException
 * @throws InterruptedException
 */
protected boolean verifyJmsMessageReceivedOnChannel(Object obj, SubscribableChannel expectedOutputChannel, CountDownHandler handler) throws JMSException, InterruptedException{
    return verifyJmsMessageOnOutputChannel(obj, expectedOutputChannel, handler, 2000);
}

/**
 * Provide a message via a mock JMS template and wait for the specified timeout to receive the message on the expected channel
 * @param obj The message provided to the poller (currently must be a String)
 * @param expectedOutputChannel The expected output channel
 * @param handler An instance of CountDownHandler to handle (verify) the output message
 * @param timeoutMillisec The timeout period. Note that this must allow at least enough time to process the entire flow. Only set if the default is
 * not long enough
 * @return true if the message was received on the expected channel
 * @throws JMSException
 * @throws InterruptedException
 */
protected boolean verifyJmsMessageOnOutputChannel(Object obj, SubscribableChannel expectedOutputChannel, CountDownHandler handler,int timeoutMillisec) throws JMSException,
        InterruptedException {

    if (!(obj instanceof String)) {
        throw new IllegalArgumentException("Only TextMessage is currently supported");
    }

    /*
     * Use mocks to create a message returned to the JMS inbound adapter. It is assumed that the JmsTemplate
     * is also a mock.
     */

    TextMessage message = mock(TextMessage.class);
    doReturn(new SimpleMessageConverter()).when(jmsTemplate).getMessageConverter();
    doReturn(message).when(jmsTemplate).receiveSelected(anyString());

    String text = (String) obj;

    CountDownLatch latch = new CountDownLatch(1);
    handler.setLatch(latch);

    doReturn(text).when(message).getText();

    expectedOutputChannel.subscribe(handler);

    boolean latchCountedToZero = latch.await(timeoutMillisec, TimeUnit.MILLISECONDS);

    if (!latchCountedToZero) {
        LOGGER.warn(String.format("The specified waiting time of the latch (%s ms) elapsed.", timeoutMillisec));
    }

    return latchCountedToZero;
}

/*
 * A MessageHandler that uses a CountDownLatch to synchronize with the calling thread
 */
private abstract class CountDownHandler implements MessageHandler {

    CountDownLatch latch;

    public final void setLatch(CountDownLatch latch){
        this.latch = latch;
    }

    protected abstract void verifyMessage(Message<?> message);

    /*
     * (non-Javadoc)
     *
     * @see
     * org.springframework.integration.core.MessageHandler#handleMessage
     * (org.springframework.integration.Message)
     */
    public void handleMessage(Message<?> message) throws MessagingException {
        verifyMessage(message);
        latch.countDown();
    }
}

但我得到以下异常:

[0;33mWARN  [main] [InboundChannelFlowUnitTest] The specified waiting time of the latch (2000 ms) elapsed.
[m
java.lang.AssertionError: message not sent to expected output channel

有什么暗示吗?

编辑:

我添加了以下测试:

    @SuppressWarnings("unchecked")
    @Test
    public void testMessageDriven() throws Exception {
        TextMessage message = mock(TextMessage.class);
        when(message.getText()).thenReturn("foo");
        Session session = mock(Session.class);
        ((SessionAwareMessageListener<TextMessage>) this.messageListenerContainer.getMessageListener()).onMessage(message, session);
        CountDownHandler myCountDownHandler = new CountDownHandler() {
            @Override
            protected void verifyMessage(Message<?> message) {
                assertNotNull(message);
                assertEquals("hello", message.getPayload());
            }
        };
        CountDownLatch myLatch = new CountDownLatch(2);
        myCountDownHandler.setLatch(myLatch);
        this.inbound.subscribe(myCountDownHandler);

        boolean receivedBeforeZero = myLatch.await(3, TimeUnit.SECONDS);

        assertTrue(receivedBeforeZero);
    }

并将消息驱动的适配器更改为:

<int-jms:message-driven-channel-adapter id="jmsIn"
                                            channel="inbound"
                                            container="messageListenerContainer"
                                            acknowledge="transacted"/>

但仍然会收到以下错误:

[0;33mWARN  [main] [InboundChannelFlowUnitTest] The specified waiting time of the latch (3 sec) elapsed.
[m
java.lang.AssertionError
    at org.junit.Assert.fail(Assert.java:92)
    at org.junit.Assert.assertTrue(Assert.java:43)
    at org.junit.Assert.assertTrue(Assert.java:54)
答案

消息驱动的适配器不使用JmsTemplate,因此模拟它,它的接收方法将不会做任何事情。

您必须模拟/存根消息侦听器容器并调用其MessageListener。您可以通过'container'属性将模拟容器提供给适配器。

编辑:

为什么你需要模拟/测试框架组件并不完全清楚;您只需将测试消息发送到流中即可。

但是,如果您使用自定义消息转换器,并且想要就地测试它,则可以模拟容器。

Here's how to do it

以上是关于测试JMS和Spring集成的主要内容,如果未能解决你的问题,请参考以下文章

sping揭秘24Spring框架对JMS的集成(无环境版,以后学MQ的时候再隆重介绍)& 任务调度和线程池

通过 Spring 集成入站适配器接收 JMS 消息随机失败

Spring集成:控制ActiveMQ连接

在使用 JMS 消息之前保持 spring 上下文处于活动状态

使用Spring集成组件关联2个JMS队列之间的消息

调度程序没有订阅者-spring集成JMS