弹簧集成单元测试出站通道适配器

Question

我有以下配置

<int:channel id="notificationChannel" datatype="com.mycompany.integration.NotificationMessage"> 
     <int:queue message-store="jdbc-message-store" capacity="1000" /> 
    </int:channel> 

    <int:outbound-channel-adapter ref="notificationHandler" 
     method="handle" channel="notificationChannel" > 
     <int:poller max-messages-per-poll="100" fixed-delay="60000" 
      time-unit="MILLISECONDS" > 
      <int:transactional isolation="DEFAULT" /> 
     </int:poller> 
    </int:outbound-channel-adapter> 

现在我想单元测试这一点，我需要等待消息被在测试正确处理，我尝试了与拦截器，但不起作用，因为我只能同步消息传递，但不能成功处理消息。在处理完成时实现发送回复，但这意味着只会执行此操作以使我的单元测试工作，在生产中，消息头中不会设置replyChannel。我怎么能实现同步成功处理的请求，而不是在messageHandler中实现它？

Answer 1

如果你正在使用Spring Integration的2.2.x的，您可以用建议这么做...

public class CompletionAdvice extends AbstractRequestHandlerAdvice { 

    private final CountDownLatch latch = new CountDownLatch(1); 

    @Override 
    protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception { 
     Object result = callback.execute(); 
     latch.countDown(); 
     return result; 
    } 

    public CountDownLatch getLatch() { 
     return latch; 
    } 

} 

在您的测试环境，建议用bean工厂后处理器添加到适配器的处理程序。

public class AddCompletionAdvice implements BeanFactoryPostProcessor { 

    private final Collection<String> handlers; 

    private final Collection<String> replyProducingHandlers; 

    public AddCompletionAdvice(Collection<String> handlers, Collection<String> replyProducingHandlers) { 
     this.handlers = handlers; 
     this.replyProducingHandlers = replyProducingHandlers; 
    } 

    @Override 
    public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException { 
     for (String beanName : handlers) { 
      defineAdviceAndInject(beanFactory, beanName, beanName + "CompletionAdvice"); 
     } 
     for (String beanName : replyProducingHandlers) { 
      String handlerBeanName = beanFactory.getAliases(beanName + ".handler")[0]; 
      defineAdviceAndInject(beanFactory, handlerBeanName, beanName + "CompletionAdvice"); 
     } 
    } 

    private void defineAdviceAndInject(ConfigurableListableBeanFactory beanFactory, String beanName, String adviceBeanName) { 
     BeanDefinition serviceHandler = beanFactory.getBeanDefinition(beanName); 
     BeanDefinition advice = new RootBeanDefinition(CompletionAdvice.class); 
     ((BeanDefinitionRegistry) beanFactory).registerBeanDefinition(adviceBeanName, advice); 
     serviceHandler.getPropertyValues().add("adviceChain", new RuntimeBeanReference(adviceBeanName)); 
    } 

} 

后处理器添加到配置<bean class="foo.AddCompletionAdvice" />。

最后，注入的意见（S）到你的测试用例

@ContextConfiguration 
@RunWith(SpringJUnit4ClassRunner.class) 
public class TestAdvice { 

    @Autowired 
    private CompletionAdvice fooCompletionAdvice; 

    @Autowired 
    private CompletionAdvice barCompletionAdvice; 

    @Autowired 
    private MessageChannel input; 

    @Test 
    public void test() throws Exception { 
     Message<?> message = new GenericMessage<String>("Hello, world!"); 
     input.send(message); 
     assertTrue(fooCompletionAdvice.getLatch().await(1, TimeUnit.SECONDS)); 
     assertTrue(barCompletionAdvice.getLatch().await(1, TimeUnit.SECONDS)); 
    } 

} 

，等待锁存器（ES）。

<int:publish-subscribe-channel id="input"/> 

<int:outbound-channel-adapter id="foo" channel="input" ref="x" method="handle"/> 

<int:service-activator id="bar" input-channel="input" ref="x"/> 

<bean class="foo.AddCompletionAdvice"> 
    <constructor-arg name="handlers"> 
     <list> 
      <value>foo</value> 
     </list> 
    </constructor-arg> 
    <constructor-arg name="replyProducingHandlers"> 
     <list> 
      <value>bar</value> 
     </list> 
    </constructor-arg> 
</bean> 

<bean id="x" class="foo.Foo" /> 

我添加这些类到Gist

编辑：更新，以提供最终消费者一般情况下（无应答），并回复产生的消费者。