Register a generic handler in SourcePollingChannelAdapter

Question

I have a simple integration flow and I'm using SourcePollingChannelAdapter for integration testing.

This is the flow:

@Bean
public MessageSource<LatestBook> booksSource() {
    return () -> MessageBuilder.withPayload(booksService.getBook()).build();
}

@Bean
public IntegrationFlow syncBooks(Trigger trigger) {
    return IntegrationFlow
            .from(booksSource(), c -> c.poller(Pollers.trigger(trigger))
                    .autoStartup(false)
                    .id(BOOKS_POLLING_ADAPTER_ID))
            .handle(new BooksHandler(booksRepository))
            .channel(syncBooksMessageChannel())
            .get();
}

@Bean
@Qualifier(SYNC_BOOKS_MESSAGE_CHANNEL)
public MessageChannel syncBooksMessageChannel(){
    return Kafka.channel(kafkaTemplate, containerFactory, kafkaTemplate.getDefaultTopic())
            .id(SYNC_BOOKS_MESSAGE_CHANNEL).get();
}

... // Other flows

And this is the test environment:

@Autowired
@Qualifier(BOOKS_POLLING_ADAPTER_ID)
protected SourcePollingChannelAdapter booksPollingSource;

@Autowired
@Qualifier(SYNC_BOOKS_MESSAGE_CHANNEL)
private SubscribableChannel syncBooksMessageChannel;

@Autowired
private BooksHandler booksHandler;

@Test
public void testBooks() {
    booksPollingSource.setOutputChannel(syncBooksMessageChannel);
    booksPollingSource.setAutoStartup(false);

    // booksPollingSource.addHandler(booksHandler); need something like this

    booksPollingSource.start();
}

I can indicate the OutputChannel and AutoStartup of the booksPollingSource simply, but how I can tell booksPollingSource to use BooksHandler before writing to the syncBooksMessageChannel.

Currently, the BooksHandler is not triggered, and I can see un-handled messages in the kafka channel.

Answer 1

You probably need to looks into a MockIntegration framework: https://docs.spring.io/spring-integration/docs/current/reference/html/testing.html#testing-mocks.

The point is that SourcePollingChannelAdapter produces messages from its source to the provided channel. The handler is a subscriber to that channel on the other side. Both of them don't know about each other. That's the whole point of Messaging - to have your solution as loosely coupled as possible.

I'm not sure what is your logic, but for now you are fully bypassing the flow you show with that setOutputChannel(syncBooksMessageChannel). Meaning that indeed a message produced from source goes to the Kafka topic. If you explain a bit more your requirements, we can come up with a testing ideas for you.