3

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.

3 Answers 3

3

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.

1

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

0

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

1
  • Both routes are consuming different files from different input folders, I'm sorry if it wasn't clear on my question. I've updated it. Apr 4, 2013 at 14:06

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.