Как я могу читать файлы из двух разных каталогов, используя @InboundChannelAdapter (Spring Integration с настройкой Java)

Question

Я пытаюсь реализовать следующий сценарий:

У меня есть триггерный файл и файл данных, которые хранятся в разных каталогах. Только если я получу файл триггера, я смогу получить доступ к файлу данных, а затем выполнить логику разделения и дальнейшей обработки. Кроме того, дело в том, что будет один файл триггера, но несколько файлов данных. Поэтому после получения файла триггера я смогу обработать все файлы данных.

Ниже приведен код, который я использовал, но он выбирает только один каталог

private static final Logger LOGGER = LoggerFactory.getLogger(DatastreamApplication.class);
private static final String DATA_DIRECTORY_PATH = "dataDirectoryLocation";

@SuppressWarnings("deprecation")
public static void main(String[] args) {
    new SpringApplicationBuilder(DatastreamApplication.class).web(false).run(args);
}

@Bean
@InboundChannelAdapter(channel = "fileInputChannel", poller = @Poller(fixedDelay = "5000"))
public MessageSource<File> sftpMessageSource() {
    FileReadingMessageSource source = new FileReadingMessageSource();
    source.setDirectory(new File(DATA_DIRECTORY_PATH));
    source.setFilter(new AcceptOnceFileListFilter<>());
    return source;
}

@Splitter(inputChannel = "fileInputChannel")
@Bean
public FileSplitter fileSplitter() {
   FileSplitter fileSplitter = new FileSplitter();
   fileSplitter.setOutputChannelName("chunkingChannel");
   return fileSplitter;
}

@ServiceActivator(inputChannel = "chunkingChannel")
@Bean
public AggregatingMessageHandler  chunker() {
    AggregatingMessageHandler aggregator = new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor());
    aggregator.setReleaseStrategy(new MessageCountReleaseStrategy(1000));
    aggregator.setExpireGroupsUponCompletion(true);
    aggregator.setGroupTimeoutExpression(new ValueExpression<>(100L));
    aggregator.setSendPartialResultOnExpiry(true);
    aggregator.setOutputChannelName("processFileChannel");
    return aggregator;
}

@Bean
@ServiceActivator(inputChannel = "processFileChannel")
public MessageHandler handler() {
    return new MessageHandler() {

        @Override
        public void handleMessage(Message<?> message) throws MessagingException {
            List<String> strings = (List<String>) message.getPayload();
            System.out.println( "List Size :  "+ strings.size() + " for List " + strings.toString());
        }

    };
}

Answer 1

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

EDIT

@SpringBootApplication
public class So52231415Application {

    public static void main(String[] args) {
        new File("/tmp/bar/baz1.txt").delete();
        new File("/tmp/bar/baz2.txt").delete();
        new File("/tmp/foo/baz.trigger").delete();
        SpringApplication.run(So52231415Application.class, args);
    }

    @Bean
    public IntegrationFlow flow() {
        return IntegrationFlows.from(Files.inboundAdapter(new File("/tmp/foo"))
                    .filterExpression("name.endsWith('.trigger')"), e -> e.poller(Pollers.fixedDelay(5_000)))
                .<File, List<File>>transform(f -> {
                    File[] files = new File("/tmp/bar").listFiles();
                    String prefix = f.getName().substring(0, f.getName().lastIndexOf('.'));
                    return Arrays.stream(files)
                        .filter(ff -> ff.getName().startsWith(prefix))
                        .collect(Collectors.toList());
                })
                .split()
                .headerFilter(FileHeaders.ORIGINAL_FILE, FileHeaders.FILENAME, FileHeaders.RELATIVE_PATH)
                .split(new FileSplitter())
                .handle(System.out::println)
                .get();
    }

    @Bean
    public ApplicationRunner runner() {
        return args -> {
            FileOutputStream fos = new FileOutputStream(new File("/tmp/bar/baz1.txt"));
            fos.write("one\ntwo\nthree\n".getBytes());
            fos.close();
            fos = new FileOutputStream(new File("/tmp/bar/baz2.txt"));
            fos.write("four\nfive\nsix\n".getBytes());
            fos.close();
            fos = new FileOutputStream(new File("/tmp/foo/baz.trigger"));
            fos.write("\n".getBytes());
            fos.close();
        };
    }

}

и

GenericMessage [payload=one, headers={sequenceNumber=1, sequenceDetails=[[a06d78dc-8a29-34aa-d8c1-64468edded5b, 1, 2]], sequenceSize=0, file_name=baz1.txt, correlationId=6581c159-352a-b943-d325-db58761b573d, file_originalFile=/tmp/bar/baz1.txt, id=6663d227-ee87-c997-6c67-01b4e50f7b6a, timestamp=1538584697084}]
GenericMessage [payload=two, headers={sequenceNumber=2, sequenceDetails=[[a06d78dc-8a29-34aa-d8c1-64468edded5b, 1, 2]], sequenceSize=0, file_name=baz1.txt, correlationId=6581c159-352a-b943-d325-db58761b573d, file_originalFile=/tmp/bar/baz1.txt, id=d48caf6e-5f3c-76cc-98a8-63800ee1acb8, timestamp=1538584697084}]
GenericMessage [payload=three, headers={sequenceNumber=3, sequenceDetails=[[a06d78dc-8a29-34aa-d8c1-64468edded5b, 1, 2]], sequenceSize=0, file_name=baz1.txt, correlationId=6581c159-352a-b943-d325-db58761b573d, file_originalFile=/tmp/bar/baz1.txt, id=8486ed52-d6a3-6eaf-20de-555b18ea0d75, timestamp=1538584697084}]
GenericMessage [payload=four, headers={sequenceNumber=1, sequenceDetails=[[a06d78dc-8a29-34aa-d8c1-64468edded5b, 2, 2]], sequenceSize=0, file_name=baz2.txt, correlationId=5842bfd7-f25c-9faf-e2f8-7b2a2badf869, file_originalFile=/tmp/bar/baz2.txt, id=3d301ccf-2b80-d4a4-4c59-5f9e38180fe5, timestamp=1538584697087}]
GenericMessage [payload=five, headers={sequenceNumber=2, sequenceDetails=[[a06d78dc-8a29-34aa-d8c1-64468edded5b, 2, 2]], sequenceSize=0, file_name=baz2.txt, correlationId=5842bfd7-f25c-9faf-e2f8-7b2a2badf869, file_originalFile=/tmp/bar/baz2.txt, id=4e43f02c-d64e-0d11-18a4-3ab8b6e7d383, timestamp=1538584697087}]