如何在 JUnit 中等待 @JMSListener 注解的方法完成

Question

所以我正在尝试对基于 Spring (v4.1.6) 的代码的 JMS 处理进行一些集成测试。

这是一个非常标准的 Spring 设置，带有 @JmsListener 注释方法和带有 concurrency 设置为 1 的 DefaultMessageListenerContainer，因此只允许 1 个侦听线程。

现在，我利用 ActiveMQ 的嵌入式代理不依赖任何外部 jms 代理来让测试随时随地运行（我应该从事营销工作）。

所以一切正常，然后我进行了 JUnit 测试：

@Test
public void test() {
    sendSomeMessage();
    //how to wait here for the @JMSListener method to complete
    verify();
}

我发送了消息，但是我需要以某种方式等待@JMSListener 带注释的方法完成。我该怎么做？

Answer 1

好吧，我希望我能以某种方式连接到消息驱动 Pojos 生命周期来执行此操作，但是通过其他关于异步代码的 SO 问题我想出了一个基于 CountDownLatch 的解决方案

1. 所有工作完成后，@JMSListener 注释方法应在 CountDownLatch 上调用 countDown()：

   @JmsListener(destination = "dest", containerFactory = "cf")
   public void processMessage(TextMessage message) throws JMSException {
       //do the actual processing
       actualProcessing(message);
       //if there's countDownLatch call the countdown.
       if(countDownLatch != null) {
           countDownLatch.countDown();
       }
   }
   

2. 在测试方法中

   @Test
   public void test() throws InterruptedException {
       //initialize the countDownLatch and set in on the processing class
       CountDownLatch countDownLatch = new CountDownLatch(1);
       messageProcessor.setCountDownLatch(countDownLatch);
       //sendthemessage
       sendSomeMessage();
       //wait for the processing method to call countdown()
       countDownLatch.await();
       verify();
   }
   

此解决方案的缺点是您必须实际更改您的 @JMSListener 注释方法，专门用于集成测试

Answer 2

为了避免更改您的实际@JmsListener 方法，您可以尝试在测试中使用AOP...

首先像这样创建一个方面类：

@Aspect
public static class JmsListenerInterceptor {
    @org.aspectj.lang.annotation.After("@annotation(org.springframework.jms.annotation.JmsListener)")
    public void afterOnMessage(JoinPoint jp) {
        // Do countdown latch stuff...
    }
}

然后将其添加到您用于测试的应用程序上下文配置中，如下所示：

<aop:aspectj-autoproxy/>
<bean id="jmsListenerInterceptor" class="path.to.your.Test$JmsListenerInterceptor" />

如果一切按计划进行，JmsListenerInterceptor 将倒计时，您无需更改实际代码。

重要提示：我刚刚发现使用 AOP 和 Mockito 来验证您的 @JmsListener 中的某些方法是否已被调用是一个糟糕的组合。原因似乎是对 CGLib 类的额外包装导致调用错误/实际目标实例而不是 Mockito 代理。

在我的测试中，我有一个@Autowired、@InjectMocks Listener 对象和一个@Mock Facade 对象，我想验证是否调用了某个方法.

使用 AOP：

• 测试线程：
  • [JmsListenerTest] 2279812 - 类 Listener$$EnhancerBySpringCGLIB$$6587f46b （由 Spring AOP 包装）
  • [JmsListenerTest] 30960534 - 类 Facade$$EnhancerByMockitoWithCGLIB$69fe8952 （由 Mockito 包装）
• 监听线程：
  • [Listener] 1151375 - 类监听器（AOP 包装类的目标实例）
  • [Listener] 4007155 - 类 FacadeImpl （不是我们预期的实际实例）

没有 AOP：

• 测试线程：
  • [JmsListenerTest] 10692528 - 类监听器（实际实例）
  • [JmsListenerTest] 823767 - 类 Facade$$EnhancerByMockitoWithCGLIB$$773538e8 （由 Mockito 包装）
• 监听线程：
  • [Listener] 10692528 - 类监听器（仍然是实际实例）
  • [Listener] 823767 - 类 Facade$$EnhancerByMockitoWithCGLIB$$773538e8 （仍然是我们的模拟实例）

这表明您需要像我尝试的那样小心使用 AOP，因为您最终可能会在两个线程中使用不同的实例...

Answer 3

如果您要将日志记录添加到 @JmsListener 注释方法中，您可以在测试类中执行类似的操作

@Rule
public OutputCapture outputCapture = new OutputCapture();

@Test
public void test() {
    sendSomeMessage();
    //how to wait here for the @JMSListener method to complete
    Assertions.assertThat(outputCapture.toString()).contains("Message received.");
}

Answer 4

我使用弹簧配置文件，并且在测试和生产代码中有不同的Processor。在我的测试代码中，我在处理后写入BlockingQueue，可以在测试中等待

例如：

@Configuration
public class MyConfiguration {
   @Bean @Profile("!test")
   public Processor productionProcessor() {
      return new ProductionProcessor();
   }
   @Bean @Profile("test")
   public Processor testProcessor() {
      return new TestProcessor();
   }
   @Bean
   public MyListener myListener(Processor processor) {
      return new MyListener(processor);
   }
}

public class MyListener {
   private final Processor processor;
   // constructor
   @JmsListener(destination = "dest", containerFactory = "cf")
   public void processMessage(TextMessage message) throws JMSException {
      processor.process(message);
   }
}

public class TestProcessor extends ProductionProcessor {
   private final BlockingQueue<TextMessage> queue = new LinkedBlockingQueue<>();
   public void process(Textmessage message) {
      super.process(message);
      queue.add(message);
   }
   public BlockingQueue getQueue() { return queue; }
}

@SpringBootTest
@ActiveProfiles("test")
public class MyListenerTest {
   @Autowired
   private TestProcessor processor;

   @Test
   public void test() {
      sendTestMessageOverMq();
      TextMessage processedMessage = processor.getQueue().poll(10, TimeUnit.SECONDS);
      assertAllOk(processedMessage);
   }

}