阿帕奇骆驼聚合完成不起作用

Question

我已经配置了一个路由来从交易所中提取一些数据并聚合它们;这是简单的总结：

@Component
@RequiredArgsConstructor
public class FingerprintHistoryRouteBuilder extends RouteBuilder {

    private final FingerprintHistoryService fingerprintHistoryService;

    @Override
    public void configure() throws Exception {
        from("seda:httpFingerprint")
                .aggregate( (AggregationStrategy) (oldExchange, newExchange) -> {
                    final FingerprintHistory newFingerprint = extract(newExchange);
                    if (oldExchange == null) {
                        List<FingerprintHistory> fingerprintHistories = new ArrayList<>();
                        fingerprintHistories.add(newFingerprint);
                        newExchange.getMessage().setBody(fingerprintHistories);
                        return newExchange;
                    }

                    final Message oldMessage = oldExchange.getMessage();
                    final List<FingerprintHistory> fingerprintHistories = (List<FingerprintHistory>) oldMessage.getBody(List.class);
                    fingerprintHistories.add(newFingerprint);

                    return oldExchange;
                })
                .constant(true)
                .completionSize(aggregateCount)
                .completionInterval(aggregateDuration.toMillis())
                .to("direct:processFingerprint")
                .end();

        from("direct:processFingerprint")
                .process(exchange -> {
                    List<FingerprintHistory> fingerprintHistories = exchange.getMessage().getBody(List.class);
                    fingerprintHistoryService.saveAll(fingerprintHistories);
                });
strong text
    }

}

问题是聚合完成永远不起作用，例如，这是我的测试示例：

@SpringBootTest
class FingerprintHistoryRouteBuilderTest {

    @Autowired
    ProducerTemplate producerTemplate;

    @Autowired
    FingerprintHistoryRouteBuilder fingerprintHistoryRouteBuilder;

    @Autowired
    CamelContext camelContext;

    @MockBean
    FingerprintHistoryService historyService;

    @Test
    void api_whenAggregate() {
        UserSearchActivity activity = ActivityFactory.buildSampleSearchActivity("127.0.0.1", "salam", "finger");
        Exchange exchange = buildExchange();
        exchange.getMessage().setBody(activity);

ReflelctionTestUtils.setField；ReflectionTestUtils.setFiled；producerTemplate.send（FingerprintHistoryRouteBuilder.FINGERPRINT_HISTORY_ENDPOINT，交换）；Mockito.verify（历史服务）. saveAll（Mockito.any ()); }

    Exchange buildExchange() {
        DefaultExchange defaultExchange = new DefaultExchange(camelContext);
        defaultExchange.setMessage(new DefaultMessage(camelContext));
        return defaultExchange;
    }

}

结果如下：

需要但未调用:fingerprintHistoryService bean . save all()；

Answer 1

我构建了这个简化的示例，测试通过了，所以看起来您对聚合的使用可能是正确的。

您是否考虑过您的 Mockito.verify（） 呼叫是在交换完成路由之前发生的？您可以通过删除验证调用并将.log（）语句添加到FINGERPRINT_PROCESS_AGGREGATION路由来测试这一点。如果在执行期间看到日志输出，则表示交换正在按预期进行路由。如果是这种情况，则您的 verify（） 呼叫需要能够等待交换完成路由。我不怎么使用模拟，但看起来你可以这样做：

Mockito.verify(historyService, timeout(10000)).saveAll(Mockito.any());