Как динамически регистрировать входящий входящий адаптер в Spring Integration?

Question

Я пытаюсь реализовать агрегатор подачи RSS/Atom в spring-integration, и я в основном использую Java DSL для написания своего IntegrationFlow. Требование этого агрегатора заключается в том, что каналы могут быть добавлены/удалены во время выполнения. То есть, каналы не известны во время разработки.

Мне было просто использовать базовый Feed.inboundAdapter() с тестовым URL-адресом и извлечь ссылки из фида с помощью трансформатора, а затем передать его на outbound-file-adapter, чтобы сохранить ссылки на файл. Тем не менее, я очень сильно застрял, пытаясь прочитать (тысячи) корневых URL-адресов из файла inbound-file-adapter, запустив файл через FileSplitter, а затем передайте каждый полученный Message<String>, содержащий URL-адрес канала, чтобы зарегистрировать новый Feed.inboundAdapter(). Не возможно ли это с помощью DSL Java?

В идеале я хотел бы, если бы я мог сделать следующее:

@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(); 
} 

Хотя после прочтения через несколько раз весны интеграции Java DSL коды (и обучение тонны материала по пути) Я просто не могу» видеть, что это можно сделать так. Итак ... А) не так ли? B) должно ли это быть? C) Предложения?

Почти похоже на то, что я должен принимать вывод .handle(new FileSplitter()) и передавать его в .handleWithAdapter(Feed.inboundAdapter(/*stuff here*/)), но DSL ссылается только на outbound-adapter. Входящие адаптеры на самом деле просто подкласс AbstractMessageSource, и кажется только место, которое вы можете указать, один из них является аргументом метода IntegrationFlows.from(/*stuff here*/).

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

Есть ли какие-то умные подклассы, которые я могу сделать, чтобы сделать эту работу?

В противном случае ... и я подозреваю, что это будет ответ, я нашел интеграцию пружинного Dynamic Ftp Channel Resolver Example и this ответ о том, как адаптировать его динамически регистрировать материал для въездного случае ...

Так это путь? Любая помощь/руководство оценены. После того, как вы пролили код DSL и прочитали документацию в течение нескольких дней, я думаю, что я буду участвовать в реализации динамического примера ftp и адаптирую его для работы с FeedEntryMessageSource ... в этом случае мой вопрос ... что работает динамический пример ftp с конфигурацией XML, но возможно ли это сделать либо с помощью Java config, либо с DSL Java?

Update

Я реализовал решение следующим образом:

@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 У меня есть o ne URL-адресов, определенных в application.properties как feed.url=[insert url here].В противном случае он не сказал мне «неспособный разрешить свойство {feed.url}». Я подозреваю, что здесь происходит то, что @Bean s, определенные в DynamicFeedAdapter.java, все получают синглтоны, нетерпеливо инициализированные, поэтому в стороне от того, что бобы создаются вручную в нашем цикле for в основном методе (который отлично работает, потому что у них есть свойство feed.url), мы иметь бездомный синглтон, который был нетерпеливо инициализирован, и если нет feed.url, определенному в application.properties, тогда он не может разрешить свойство, и все идет ударом. Теперь из того, что я знаю о Spring, я знаю, что должно быть возможно, чтобы @Lazy инициализировал компоненты в DynamicFeedAdapter.java, поэтому мы не завершаем эту проблему нежелательной случайной проблемой singleton. Проблема в том, что ... если я просто отметю feedAdapter()@Lazy, тогда бобы никогда не будут инициализированы. Как инициализировать их самостоятельно?

Update - проблема решена

Без испытав его, я думаю, что проблема в том, что загрузочный найти DynamicFeedAdapter во время его компоненты сканирования. Простым решением является , чтобы переместить его в пакет для родственников. Если MonsterFeedApplication находится в com.acme.foo, тогда установите класс конфигурации адаптера в com.acme.bar. То, что способ, ботинок не будет считаться «частью» приложения

Это действительно была проблема. После реализации предложения Гэри все работает отлично.

Answer 1

См. the answer to this question и its follow up по аналогичному вопросу о адаптерах входящей почты.

По существу, каждый адаптер подачи создается в дочернем контексте, который параметризуется.

В этом случае дочерние контексты создаются в методе main(), но нет причин, по которым это невозможно сделать в службе, вызванной .handle().