使用 Spring Cloud Stream Test 进行错误测试

Question

我们使用spring-cloud-stream 来管理我们的应用程序之间的消息。

我们有自定义绑定：

public interface InboundChannels {

  String TASKS = "domainTasksInboundChannel";
  String EVENTS = "eventsInboundChannel";

  @Input(TASKS)
  SubscribableChannel tasks();

  @Input(EVENTS)
  SubscribableChannel events();

}

public interface OutboundChannels {

  String TASKS = "domainTasksOutboundChannel";
  String EVENTS = "eventsOutboundChannel";

  @Output(TASKS)
  MessageChannel tasks();

  @Output(EVENTS)
  MessageChannel events();

}

有处理器消耗任务并生成事件：

@EnableBinding({InboundChannels.class, OutboundChannels.class})
public class TasksProcessor {

  public TasksProcessor(
        UserService userService,
        @Qualifier(OutboundChannels.EVENTS) MessageChannel eventsChannel
  ) {
    this.userService = userService;
    this.eventsChannel = eventsChannel;
  }

  @StreamListener(value = TASKS, condition = "headers['" + TYPE + "']=='" + CREATE_USER + "'")
  public void createUser(Message<User> message)  {
    final User user = message.getPayload();
    userService.save(user)
            .subscribe(created -> {
                Message<User> successMessage = fromMessage(message, Events.USER_CREATED, created).build();
                eventsChannel.send(successMessage);
            });
  }

}

现在我们想使用spring-cloud-stream-test-support 及其惊人的功能对其进行测试：

@DirtiesContext
@SpringBootTest
@RunWith(SpringRunner.class)
public class TasksProcessorTest {

  private User user;

  @Autowired
  private InboundChannels inboundChannels;

  @Autowired
  private OutboundChannels outboundChannels;

  @Autowired
  private MessageCollector collector;

  @Before
  public void setup() {
    user = new User(BigInteger.ONE, "test@teste.com");
  }

  @Test
  public void createUserTest() {
    final Message<User> msg = create(CREATE_USER, user).build();
    outboundChannels.tasks().send(msg);
    final Message<?> incomingEvent = collector.forChannel(inboundChannels.events()).poll();
    final String type = (String) incomingEvent.getHeaders().get(TYPE);
    assertThat(type).isEqualToIgnoringCase(USER_CREATED);
  }

}

application.properties

##
# Spring AMQP configuration
##
spring.rabbitmq.host=rabbitmq
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin

# Events channels
spring.cloud.stream.bindings.eventsOutboundChannel.destination=events
spring.cloud.stream.bindings.eventsInboundChannel.destination=events

spring.cloud.stream.bindings.domainTasksOutboundChannel.destination=domainTasks

spring.cloud.stream.bindings.domainTasksInboundChannel.destination=domainTasks

spring.cloud.stream.bindings.userTasksInboundChannel.group=domainServiceInstances spring.cloud.stream.bindings.eventsInboundChannel.group=domainServiceInstances

然后我们得到这个错误：

java.lang.IllegalArgumentException: Channel [eventsInboundChannel] was not bound by class org.springframework.cloud.stream.test.binder.TestSupportBinder

我们做错了什么？

Answer 1

在.subscribe() 中，您执行eventsChannel.send(successMessage);，其中eventsChannel 来自OutboundChannels.EVENTS，但您在测试用例中尝试执行的操作类似于inboundChannels.events()。而且看起来你并没有真正将这个频道绑定到任何地方。

我确定您是否会改用 outboundChannels.events()，这对您有用。