spring-интеграция-dsl: сделайте работу подачи потока

Question

Я пытаюсь закодировать программу чтения RSS-каналов с помощью настроенного набора RSS-каналов. Я подумал, что хороший подход - это решить путем кодирования прототипа@Bean и вызывайте его с каждым RSS-каналом, найденным в конфигурации.

Тем не менее, я предполагаю, что мне не хватает точки здесь при запуске приложения, но ничего не происходит. Я имею в виду, что бобы создаются так, как я ожидал, но в этом не происходит никакой регистрации handle()-метод:

@Component
public class HomeServerRunner implements ApplicationRunner {

    private static final Logger logger = LoggerFactory.getLogger(HomeServerRunner.class);

    @Autowired
    private Configuration configuration;

    @Autowired
    private FeedConfigurator feedConfigurator;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        List<IntegrationFlow> feedFlows = configuration.getRssFeeds()
            .entrySet()
            .stream()
            .peek(entry -> System.out.println(entry.getKey()))
            .map(entry -> feedConfigurator.feedFlow(entry.getKey(), entry.getValue()))
            .collect(Collectors.toList());
        // this one appears in the log-file and looks good
        logger.info("Flows: " + feedFlows);
    }

}

@Configuration
public class FeedConfigurator {

    private static final Logger logger = LoggerFactory.getLogger(FeedConfigurator.class);

    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public IntegrationFlow feedFlow(String name, FeedConfiguration configuration) {
        return IntegrationFlows
                .from(Feed
                        .inboundAdapter(configuration.getSource(), getElementName(name, "adapter"))
                        .feedFetcher(new HttpClientFeedFetcher()),
                        spec -> spec.poller(Pollers.fixedRate(configuration.getInterval())))
                .channel(MessageChannels.direct(getElementName(name, "in")))
                .enrichHeaders(spec -> spec.header("feedSource", configuration))
                .channel(getElementName(name, "handle"))
        //
        // it would be nice if the following would show something:
        //
                .handle(m -> logger.debug("Payload: " + m.getPayload()))
                .get();
    }

    private String getElementName(String name, String postfix) {
        name = "feedChannel" + StringUtils.capitalize(name);
        if (!StringUtils.isEmpty(postfix)) {
            name += "." + postfix;
        }
        return name;
    }

}

Чего здесь не хватает? Кажется, мне нужно как-то "запустить" потоки.

Answer 1

Бобы прототипа нужно где-то "использовать" - если вы нигде не ссылаетесь на них, экземпляр не будет создан.

Кроме того, вы не можете положить IntegrationFlow@Bean в этой области - он генерирует внутреннюю группу бинов, которых не будет в этой области.

Посмотрите ответ на этот вопрос и его продолжение для одного метода, который вы можете использовать для создания нескольких адаптеров с различными свойствами.

Кроме того, в следующей версии DSL 1.2 есть механизм динамической регистрации потоков.