Spring batch job should be executed only once after the spring integration file poller polls the files

Question

I am trying to poll the files from system folder where one or files may come, for those files, I have to trigger the batch job only once not the number of times equals to the number of files in the folder. In my case my batch is taking care the multiple files at a time, i just want poller to send the signal to batch only once to start its job.

Tried poller.maxMessagesPerPoll(1) etc but its something different.. I am facing the issue where the batch job is getting triggered equals to the number of files the poller gets in the polling folder. I want to execute batch only once

@Bean
public FileMessageToJobRequest fileMessageToJobRequest() {
    FileMessageToJobRequest fileMessageToJobRequest = new FileMessageToJobRequest();
    fileMessageToJobRequest.setJob(fileMessageBatchJob);
    return fileMessageToJobRequest;
}
@Bean
public JobLaunchingGateway jobLaunchingGateway() {
    SimpleJobLauncher simpleJobLauncher = new SimpleJobLauncher();
    simpleJobLauncher.setJobRepository(jobRepository);
    simpleJobLauncher.setTaskExecutor(new SyncTaskExecutor());
    JobLaunchingGateway jobLaunchingGateway = new JobLaunchingGateway(simpleJobLauncher);
    return jobLaunchingGateway;
}
@Bean
public IntegrationFlow integrationFlow(JobLaunchingGateway jobLaunchingGateway) {
    System.out.println("&&&&&&&&&&&&&&&&&&Inside Integration Flow!!!!");
    return IntegrationFlows
            .from(Files.inboundAdapter(new File("C:\\apps_data\\recv")),
                    c -> c.poller(Pollers.fixedDelay(1000).maxMessagesPerPoll(1)))
            .filter(onlyT4F2())
            .handle(fileMessageToJobRequest)
            .handle(jobLaunchingGateway)
            .log(LoggingHandler.Level.WARN, "headers.id + ': ' + payload").get();
}

@Bean
public GenericSelector<File> onlyT4F2() {
    System.out.println("@@@@@@@Inside GenericSelector of XXX");
    return new GenericSelector<File>() {
        @Override
        public boolean accept(File source) {
          return source.getName().contains("XXX");
        }
    };
}

Current behavior - When poller detects the file/files on given location the configured batch job triggers that many times. If files are 4 then batch job triggers 4 times.

Expected behavior - After file polling, the batch job should be get executed only once for any number of files. Because the batch job is processing the multiple files at a time hence multiple times execution is unnecessary.

Let me know if you will need any other information from my end. Please help needed on priority

Spring batch Job explicitly reading all the files from the polling location so here I am just using the poller to watch if the specific format of file is coming then trigger the batch job.. — Ganesh, Jul 15 '19 at 14:37

Gary Russell · Accepted Answer · 2019-07-15T17:31:15.573

0

You can use a custom FileListFilter on the inbound channel adapter which only returns one file.

.filter(myFilterThatOnlyReturnsOneFile)

EDIT

public class OnlyOneFileListFilter implements FileListFilter<File> {

    @Override
    public List<File> filterFiles(File[] files) {
        return Collections.singletonList(files[0]);
    }

}

        return IntegrationFlows
                .from(Files.inboundAdapter(new File("C:\\apps_data\\recv"))
                            .filter(new OnlyOneFileListFilter()),
                        c -> c.poller(Pollers.fixedDelay(1000).maxMessagesPerPoll(1)))
                ...

edited Jul 15 '19 at 17:31

answered Jul 15 '19 at 15:07

Gary Russell

166,535
14
146
179

Can you please give me code snippet.. I understood your logic but confused on how can I fit into Integration Flow.. – Ganesh Jul 15 '19 at 16:28
Russel - Thanks for your solution, I tried but getting below the exception and it looks filter method accepts boolean, could you please help Caused by: java.lang.IllegalArgumentException: a boolean result is required: class java.util.Collections$SingletonList at org.springframework.util.Assert.assignableCheckFailed(Assert.java:655) at org.springframework.util.Assert.isAssignable(Assert.java:586) at org.springframework.integration.filter.MessageFilter.doHandleRequestMessage(MessageFilter.java:165) – Ganesh Jul 16 '19 at 05:11
You have applied the `.filter()` to the main flow instead of to the adapter like I show in my answer. This is not a message filter like your `onlyT4F2`, it's a property of the channel adapter. – Gary Russell Jul 16 '19 at 11:25

score 0 · Answer 2 · answered Jul 16 '19 at 07:13

@Gary Russell - Issue is solved, used the GenericSelector only as below. Thanks for your help. After triggering the batch job on the first run it processes all the present files and moves it to some other folder hence I have added file.exists() and its working well as per my expectations. But I observed that after 1 hour or sometimes the polling is not happening even after providing the expected files, need your help/suggestion on the same.

@Bean
public GenericSelector<File> triggerJobOnlyOnce() {
    return new GenericSelector<File>() {
        @Override
        public boolean accept(File source) {
                if(source.getName().contains("XXX") && source.exists())
                    return true;
            return flag;
        }
    };
}

Spring batch job should be executed only once after the spring integration file poller polls the files

2 Answers2