CODEWITHSUNDEEP
apache-camel
José Fernández Ramos
José Fernández Ramos

Reputation: 5360

Camel: synchronization between parallel routes in same camel context

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.

Upvotes: 3

Views: 4422

Answers (3)

Peter Keller
Peter Keller

Reputation: 7646

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.

Upvotes: 3

Rafael
Rafael

Reputation: 2676

You can also place an "on completion" http://camel.apache.org/oncompletion.html in the first route that activates the second one.

Upvotes: 1

kinaesthesia
kinaesthesia

Reputation: 703

Apache camel File will create a lock for the file that being processed. Any other File process on this file will not pool on if there is a lock (except if you put consumer.exclusiveReadLock=false)

source :

http://camel.apache.org/file.html => URI Options => consumer.exclusiveReadLock

Upvotes: 0

Related Questions