测试Spring-Cloud-Stream服务的问题

Question

我在测试我的Spring Cloud stream服务（写入Kafka流）时遇到问题。它基于以下baeldung介绍

这是服务代码（具体省略）

@Service
@EnableBinding(Source.class)
public class KafkaWriterService {

    @SendTo(Source.OUTPUT)
    public String write(String str) {
        return str;
    }

}

这是测试

@SpringBootTest
@RunWith(SpringRunner.class)
public class KafkaWriterServiceTest {

    @Autowired
    private Source source;

    @Autowired
    private MessageCollector collector;

    @Autowired
    private KafkaWriterService service;

    @Test
    public void testMessages() {
        BlockingQueue<Message<?>> messages = collector.forChannel(source.output());
        service.write("FooBar");

        Object payload = messages.poll().getPayload();
        System.out.println(payload);
    }

非常直接，但是在执行测试时，我得到一个NullPointerException，因为轮询返回一个Null。

知道问题是什么吗？

谢谢

Answer 1

更新：

如果您只是在寻找Kafka生产者示例，请对您的服务进行以下更改：

@Service
@EnableBinding({Source.class})
public class KafkaWriterService {

    private final MessageChannel output;

    public KafkaWriterService(@Qualifier("output") MessageChannel output) {
        this.output = output;
    }

    public void write(String str) {
        output.send(MessageBuilder.withPayload(str).build());
    }
}

发送到需要通过StreamListener提供消息源。它用于转换输入消息。

在您的测试程序中，没有生成任何消息。

将代码更改为以下内容，以便在测试中生成消息：

@Service
@EnableBinding({Source.class, Sink.class})
public class KafkaWriterService {

    @StreamListener(Sink.INPUT)
    @SendTo(Source.OUTPUT)
    public String write(String str) {
        return "Transformed: " + str;
    }
}

@SpringBootTest
@RunWith(SpringRunner.class)
public class KafkaWriterServiceTest {

    @Autowired
    private Sink sink;

    @Autowired
    private Source source;

    @Autowired
    private MessageCollector collector;

    @Autowired
    private KafkaWriterService service;

    @Test
    public void testMessages() {
        BlockingQueue<Message<?>> messages = collector.forChannel(source.output());
        sink.input().send(MessageBuilder.withPayload("Hello World!").build());

        Object payload = messages.poll().getPayload();
        System.out.println(payload);
    }
}