如何在 Spring 集成中动态注册 Feed Inbound Adapter？

Question

我正在尝试在 spring-integration 中实现一个 RSS/Atom 提要聚合器，我主要使用 Java DSL 来编写我的 IntegrationFlow。此聚合器的要求是可以在运行时间内添加/删除提要。也就是说，提要在设计时未知。

我发现使用带有测试 url 的基本 Feed.inboundAdapter() 并使用转换器从提要中提取链接然后将其传递给 outbound-file-adapter 很简单将链接保存到文件。但是，当我试图通过 FileSplitter 从 inbound-file-adapter 运行文件读取（数千）提要 urls 然后传递每个结果 Message<String> 包含提要 url，然后注册一个新的 Feed.inboundAdapter()。 Java DSL 不可能做到这一点吗？

理想情况下，如果我能做到以下几点，我会喜欢它：

@Bean
public IntegrationFlow getFeedsFromFile() throws MalformedURLException {
    return IntegrationFlows.from(inboundFileChannel(), e -> e.poller(Pollers.fixedDelay(10000)))
            .handle(new FileSplitter())
            //register new Feed.inboundAdapter(payload.toString()) foreach Message<String> containing feed url coming from FileSplitter
            .transform(extractLinkFromFeedEntry())
            .handle(appendLinkToFile())
            .get();
}

虽然在多次阅读 spring 集成 java DSL 代码（并在此过程中学习了大量的东西）之后，我还是看不出这样做是可能的.所以... A) 是吗？ B) 应该是吗？ C）建议？

几乎感觉我应该能够获取 .handle(new FileSplitter()) 的输出并将其传递到 .handleWithAdapter(Feed.inboundAdapter(/*stuff here*/))，但 DSL 仅在此处引用 outbound-adapter。入站适配器实际上只是 AbstractMessageSource 的子类，似乎只有可以指定其中之一的地方是 IntegrationFlows.from(/*stuff here*/) 方法的参数。

我原以为可以从文件中获取输入，逐行拆分，使用该输出来注册入站提要适配器，轮询这些提要，从提要中提取出现的新链接，然后将它们附加到文件中。好像不是。

我可以做一些聪明的子类化来完成这项工作吗？

如果失败...我怀疑这将是答案，我找到了 spring 集成 Dynamic Ftp Channel Resolver Example and this 关于如何调整它动态注册入站案例的答案。 .

这是要走的路吗？任何 help/guidance 表示赞赏。在连续几天阅读 DSL 代码和阅读文档之后，我想我会着手实施动态 ftp 示例并使其适应 FeedEntryMessageSource ......在这种情况下，我的问题是......动态 ftp 示例适用于 XML 配置，但是否可以使用 Java 配置或 Java DSL 来实现？

更新

我已经实现了如下解决方案：

@SpringBootApplication 
class MonsterFeedApplication {

public static void main(String[] args) throws IOException {
    ConfigurableApplicationContext parent = SpringApplication.run(MonsterFeedApplication.class, args);

    parent.setId("parent");
    String[] feedUrls = {
            "https://1nichi.wordpress.com/feed/",
            "http://jcmuofficialblog.com/feed/"};

    List<ConfigurableApplicationContext> children = new ArrayList<>();
    int n = 0;
    for(String feedUrl : feedUrls) {
        AnnotationConfigApplicationContext child = new AnnotationConfigApplicationContext();
        child.setId("child" + ++n);
        children.add(child);
        child.setParent(parent);
        child.register(DynamicFeedAdapter.class);
        StandardEnvironment env = new StandardEnvironment();
        Properties props = new Properties();
        props.setProperty("feed.url", feedUrl);
        PropertiesPropertySource pps = new PropertiesPropertySource("feed", props);
        env.getPropertySources().addLast(pps);
        child.setEnvironment(env);
        child.refresh();
    }

    System.out.println("Press any key to exit...");
    System.in.read();
    for (ConfigurableApplicationContext child : children) {
        child.close();
    }
    parent.close();
}

@Bean
public IntegrationFlow aggregateFeeds() {       
    return IntegrationFlows.from("feedChannel")
            .transform(extractLinkFromFeed())
            .handle(System.out::println)
            .get();
}

@Bean
public MessageChannel feedChannel() {
    return new DirectChannel();
}

@Bean
public AbstractPayloadTransformer<SyndEntry, String> extractLinkFromFeed() {
    return new AbstractPayloadTransformer<SyndEntry, String>() {
        @Override
        protected String transformPayload(SyndEntry payload) throws Exception {
            return payload.getLink();
        }
    };

}

}

DynamicFeedAdapter.java

@Configuration
@EnableIntegration
public class DynamicFeedAdapter {

    @Value("${feed.url}")
    public String feedUrl;

    @Bean
    public static PropertySourcesPlaceholderConfigurer pspc() {
        return new PropertySourcesPlaceholderConfigurer();
    }

    @Bean
    public IntegrationFlow feedAdapter() throws MalformedURLException {

        URL url = new URL(feedUrl);

        return IntegrationFlows
                .from(s -> s.feed(url, "feedTest"), 
                        e -> e.poller(p -> p.fixedDelay(10000)))
                .channel("feedChannel")
                .get();
    }

}

这有效 IF 只有 IF 我有一个 urls 在 application.properties 中定义为 feed.url=[insert url here]。否则它不会告诉我 'unable to resolve property {feed.url}'。我怀疑正在发生的事情是 DynamicFeedAdapter.java 中定义的 @Bean 都得到了单例的急切初始化，所以除了在 main 方法的 for 循环中手动创建的 bean（工作正常，因为它们有 feed.url 属性注入）我们有一个已经急切初始化的流浪单例，如果 application.properties 中没有定义 feed.url 那么它无法解析属性一切顺利。现在根据我对 Spring 的了解，我知道应该可以 @Lazy 初始化 DynamicFeedAdapter.java 中的 bean，这样我们就不会以这个不需要的流浪单例结束 problem-child.现在的问题是...如果我只标记 feedAdapter() @Lazy 那么 bean 永远不会被初始化。我如何自己初始化它们？

更新 - 问题已解决

Without having tested it, I think the problem is that boot is finding the DynamicFeedAdapter during its component scan. A simple solution is to move it to a sibling package. If MonsterFeedApplication is in com.acme.foo, then put the adapter config class in com.acme.bar. That way, boot won't consider it "part" of the application

确实是这个问题。执行Gary的建议后，一切正常。

Answer 1

有关入站邮件适配器的类似问题，请参阅 and 。

本质上，每个提要适配器都是在参数化的子上下文中创建的。

在那种情况下，子上下文是在 main() 方法中创建的，但没有理由不能在 .handle() 调用的服务中完成。

如何在 Spring 集成中动态注册 Feed Inbound Adapter？

how to dynamically register Feed Inbound Adapter in Spring Integration?

rss

dsl

spring-integration

enterprise-integration

atom-feed