Camel - start route when multiple files in directory

Question

I want start camel route when 4 files with specific names are in directory. When file1.csv.sem, file2.csv.sem, file3.csv.sem, file4.csv.sem start camel route.

 <from uri="file:{{directory.path}}?include=file1.*.sem"/>

I expect camel route will start when these 4 files are in directory.

Answer 1

You could create timer or scheduled route that uses custom processor that checks contents of a folder using File.list() or File.listFiles(). If all the required files exist processor sets body, header or property to signal that the route should continue.

After that you can use PollEnrich to consume each of the files individually and store them to a header or property for later use.

String filePath = getContext()
    .resolvePropertyPlaceholders("{{input.path}}");

String[] requiredFiles = new String[]{
    "file1.csv.sem", 
    "file2.csv.sem", 
    "file3.csv.sem"
};

// Poll input folder every 5 seconds for required files
from("timer:timerName?period=5000")
    .routeId("multifiles")
    .process(new Processor() {
        @Override
        public void process(Exchange exchange) throws Exception {
            
            for (int i = 0; i < requiredFiles.length; i++) {
                File file = new File(filePath + File.separator + requiredFiles[i]);
                if(!file.exists()){
                    exchange.getMessage().setBody(false);
                    return;    
                }
            }
            exchange.getMessage().setBody(true);
        }  
    })
    .filter(body().isEqualTo(false))
        .log("Missing files..")
        .stop()
    .end()
    .to("direct:processFileGroup");

from("direct:processFileGroup")
    .routeId("processFileGroup")
    .setBody(constant(Arrays.asList(requiredFiles)))
    // Read required files and aggregate to a list
    .split(body(), AggregationStrategies.groupedBody())
        .log("${body}")
        .pollEnrich()
            .simple("file:{{input.path}}?fileName=${body}")
            .timeout(1000)
        .end()
    .end()
    // Do stuff with the List of Exchange(s) in body
    .process(new Processor() {
        @Override
        public void process(Exchange exchange) throws Exception {
            
            List<GenericFile> files = exchange.getMessage().getBody(List.class);
            for (int i = 0; i < files.size(); i++) {
                System.out.println(files.get(i).getFileName());
            }
        }
    });

Alternatively you could probably use file component with some custom aggregation strategy or some trickery with File-watch but those sound more complex at least for me.

This is fairly common problem so there's probably some sort of EIP to handle this more elegantly but have not stumbled upon such yet.