Camel: synchronization between parallel routes in same camel context

Question

I'm working on a camel prototype which uses two start points in the same camel context.

The first route consumes messages which are used to "configure" the application. Messages are loaded in a configuration repository through a configService bean:

    // read configuration files
    from("file:data/config?noop=true&include=.*.xml")
        .startupOrder(1)
        .to("bean:configService?method=loadConfiguration")
        .log("Configuration loaded");   

The second route implements a recipient list eip pattern, delivering a different kind of input messages to a number of recipients, which are read dinamically from the same configuration repository:

    // process some source files (using configuration)       
    from("file:data/source?noop=true")
        .startupOrder(2)
        .unmarshal()
        .to("setupProcessor") // set "recipients" header
        .recipientList(header("recipients"))

    // ...

The question that arises now is how to synchronize them, so the second route "waits" if the first is processing new data.

I'm new to Apache Camel and pretty lost on how to approach such a problem, any suggestion would be appreciated.

Answer 1

Use aggregate in combination with the possibility to start and stop routes dynamically:

from("file:data/config?noop=true&include=.*.xml")
    .id("route-config")
    .aggregate(constant(true), new MyAggregationStrategy()).completionSize(2).completionTimeout(2000)
    .process(new Processor() {
        @Override
        public void process(final Exchange exchange) throws Exception {
            exchange.getContext().startRoute("route-source");
        }
    });

from("file:data/source?noop=true&idempotent=false")
    .id("route-source")                              // the id is needed so that the route is found by the start and stop processors
    .autoStartup(false)                              // this route is only started at runtime
    .aggregate(constant(true), new MyAggregationStrategy()).completionSize(2).completionTimeout(2000)
    .setHeader("recipients", constant("direct:end")) // this would be done in a separate processor
    .recipientList(header("recipients"))
    .to("seda:shutdown");                            // shutdown asynchronously or the route would be waiting for pending exchanges

from("seda:shutdown")
    .process(new Processor() {
        @Override
        public void process(final Exchange exchange) throws Exception {
            exchange.getContext().stopRoute("route-source");
        }
    });

from("direct:end")
    .log("End");

That way, route-source is only started when route-config is completed. route-config and consequently route-source are restarted if new files are found in the config directory.