CODEWITHSUNDEEP
springspring-integration
anu pallavi
anu pallavi

Reputation: 23

Spring file Inbound adapters - priority to directories with single channels

Using single inbound channel I need to process two directories lowpriority and highprioiry but lowpriority files pick after the highpriority . Any one know how handle multiple directories in file inbound adapter with single channel ? 

@Bean
public IntegrationFlow processFileFlow() {
    return pushFileForProcess(lowDirectory
            ,"processFile"
            , "fileInputChannel");
}

private IntegrationFlow pushFileForProcess(String processDir, String methodName, String adapterName) {
    String fileProcessor = "fileProcessor";
    return IntegrationFlows
            .from(Files.inboundAdapter(Paths.get(processDir).toFile())
                            .regexFilter(FILE_PATTERN_REGEX)
                            .preventDuplicates(false),
                    e -> e.poller(Pollers.fixedDelay(j.getPollerIntervalMs())
                            .maxMessagesPerPoll(j.getPollerMaxMsgs())
                            .errorChannel("errorChannel")) // moves processed files
                            .id(adapterName))
            .handle(fileProcessor, methodName).get();
}

Upvotes: 1

Views: 203

Answers (1)

Gary Russell
Gary Russell

Reputation: 174779

Use a Smart Poller Advice to reconfigure the FileReadingMessageSource when the poll returns no files for the high priority directory.

Presumably you should reconfigure it back on each low-priority poll (successful or not).

EDIT

Example:

@SpringBootApplication
public class So53868122Application {

    private static final File HIGH = new File("/tmp/high");

    private static final File LOW = new File("/tmp/low");

    public static void main(String[] args) {
        HIGH.mkdir();
        LOW.mkdir();
        SpringApplication.run(So53868122Application.class, args);
    }

    @Bean
    public IntegrationFlow flow() {
        return IntegrationFlows.from(Files.inboundAdapter(HIGH),
                    e -> e.poller(Pollers.fixedDelay(5_000)
                            .advice(dirSwitcher())))
                .handle(System.out::println)
                .get();
    }

    @Bean
    public Advice dirSwitcher() {
        return new HighLowPollerAdvice();
    }

    public static class HighLowPollerAdvice extends AbstractMessageSourceAdvice {

        private boolean isHigh = true;

        @Override
        public Message<?> afterReceive(Message<?> result, MessageSource<?> source) {
            if (this.isHigh && result == null) {
                System.out.println("No results from " + HIGH + ", switching to " + LOW);
                this.isHigh = false;
                ((FileReadingMessageSource) source).setDirectory(LOW);
            }
            else if (!this.isHigh) {
                System.out.println("After one poll of " + LOW + " that returned "
                        + (result == null ? "no file" : result.getPayload()) + ", switching to " + HIGH);
                this.isHigh = true;
                ((FileReadingMessageSource) source).setDirectory(HIGH);
            }
            return result;
        }

    }

}

Upvotes: 1

Related Questions