Spring Cloud Stream test-binder OutputDestination 不消费事件答案

【问题标题】：Spring Cloud Stream test-binder OutputDestination does not consume eventsSpring Cloud Stream test-binder OutputDestination 不消费事件
【发布时间】：2023-01-03 22:39:09
【问题描述】：

我们使用微服务和事件驱动架构（更具体的编排）。我们使用 kafka，许多服务使用 Spring Cloud Stream 作为消息代理的抽象。

将我们的 Spring Cloud Stream 相关源升级到新的功能样式后，我们的集成测试开始出现问题。问题与将旧的 MessageCollector 替换为 OutputDestination(test-binder) 有关。

问题出现在我们的集成测试中，我们想在其中验证是否正在生成正确的事件。我们的许多服务产生一个主题并在另一个模块（相同服务）中使用它。 OutputDestination 现在在主题级别上工作，而不是像旧的 MessageCollector 在通道上工作。如果产品代码中已经有此主题的侦听器，它会导致 OutputDestination 不使用任何消息。

我创建了一个简单的项目来展示我们的问题https://github.com/dgyordanov/scs-functional-test

我们有一个简单的服务，例如：

@Service
public class OrderService {

.........

public void changeOrder() {
    // Some order changes
    streamBridge.send("orderEvents-out-0", "Test Order Change Event");
}

在另一个模块中，我们在生产代码中为这些事件设置了一个侦听器：

@Bean
public Consumer<String> orderEvents() {
    // React on order events
    return e -> System.out.println("### Order Event: " + e);
}

我想测试 changeOrder() 但没有消耗任何东西：

@Test
void orderChangedTest() {
   orderService.changeOrder();
   Message<byte[]> event = outputDestination.receive(100, "edu.events.orderEvents");
   assertNotNull(event);
}

当我们从上面运行测试时，我们看到来自 System.out.println("### Order Event: " + e); 的结果

问题是，如果我们不从测试上下文中排除 orderEvents() 侦听器，outputDestination 将永远不会收到消息，因为 orderEvents() 侦听器将首先使用它们。使用在通道级别工作的旧 MessageCollector，这是可能的。

你能帮我如何让我们的大黄瓜集成测试套件与 spring cloud stream test-binder 一起工作吗？

我们还尝试为同一主题声明另一个通道，但 outputDestination 仍然没有消耗任何内容。

【问题讨论】：

标签： spring-cloud-stream spring-cloud-stream-binder

【解决方案1】：

这是一个正确的行为，因为你有一个 Consumer 是该行的末尾，因为它什么都不产生。换句话说，您无法测试消费者执行的结果。

现在，如果我理解正确的话，您只是想验证确实调用了消费者（您的绑定是正确的）。我们当然可以引入对测试绑定器的增强，以允许您以更惯用的方式执行此操作（我刚刚为此提出了一个问题 - https://github.com/spring-cloud/spring-cloud-stream/issues/2607）。作为一种解决方法，您仍然可以通过一些反思来做到这一点。这是一个示例代码：

InputDestination inputDestination = context.getBean(InputDestination.class);
try {
    Field chField = ReflectionUtils.findField(InputDestination.class, "channels");
    chField.setAccessible(true);
    List<SubscribableChannel> channels = (List<SubscribableChannel>) chField.get(inputDestination);
    SubscribableChannel ch = channels.iterator().next(); // or more elaborate code if there are multiple bindings to find your channel
    ch.subscribe((x) -> {
        System.out.println("Second subscriber: " + x);
    });
} catch (Exception e) {
    // TODO: handle exception
}

【讨论】：