copy of Exchange during Multicast

[kenhans <ke...@hotmail.co.uk>]

according to  http://camel.apache.org/multicast.html
http://camel.apache.org/multicast.html , multicast will copy the source
Exchange and multicast each copy. I.e. changes done to the Exchange remains
local to the multicast recipient. Is there a way to make changes done to the
header visible in further processing? Particularly, I would like to see if
the Exchange.FILTER_MATCHED header has been set or not. With the
MulticastDefinition below, the header is lost before reaching aggregation.

Any thoughts are highly appreciated.

MulticastDefinition multicastDefinition = route.multicast(); 
for (Producer producer : producers) { 
  multicastDefinition 
  .filter().method(producer, "filter")   
  .bean(producer, "createEvent") 
  .to(EndPointUri.INCOMING_EVENT_QUEUE); 
} 
multicastDefinition.aggregate(constant(true) , new
MyAggregationStrategy()).completionTimeout(3000).to("log:dummy");

--
View this message in context: http://camel.465427.n5.nabble.com/copy-of-Exchange-during-Multicast-tp5714477.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Re: copy of Exchange during Multicast

[Henryk Konsek <he...@gmail.com>]

> Could Spring mess up things? Or am I missing out on something here? Is my
> config wrong?

Actually your example seems a little bit complicated for me. It's hard
for me to tell what's wrong there.

Could you describe me what is the purpose of your route? Maybe I could
suggest you then how would I achieve it.

-- 
Henryk Konsek
http://henryk-konsek.blogspot.com

Re: copy of Exchange during Multicast

[kenhans <ke...@hotmail.co.uk>]

Thanks, Henryk, I really appreciate your answer. 

I can see that the FilterProcessor do
exchange.setProperty(Exchange.FILTER_MATCHED, matches);
and I also see that the GroupedExchangeAggregationStrategy puts the Exchange
in it's list of answer.properties. 

I also set headers/properties myself (in bold below).

But I am still not able to access any of those headers in the
"filterMatchCheckRouter" as stated below. The headers are just not there in
the exchange sent to filterMatchCheckRouter.

Could Spring mess up things? Or am I missing out on something here? Is my
config wrong?

ProcessorDefinition processorDefinition =
from(incomingQueue).routeId("ProducerRoute").shutdownRoute(ShutdownRoute.Defer).shutdownRunningTask(ShutdownRunningTask.CompleteAllTasks).convertBodyTo(Document.class);
MulticastDefinition multicastDefinition = processorDefinition.multicast(*new
GroupedExchangeAggregationStrategy()*);
for (Producer producer : producers) { 
  multicastDefinition 
  .filter().method(producer, "filter")  
  *.process(new Processor() {
    @Override
    public void process(Exchange exchange) throws Exception {    
      exchange.getIn().setHeader("KEY1", "value1");
      exchange.setProperty("KEY2", "value2");
    }}) *
  .bean(producer, "createEvent") 
  .to(EndPointUri.INCOMING_EVENT_QUEUE); 
} 
multicastDefinition.to("direct:filterMatchCheckRouter");


--
View this message in context: http://camel.465427.n5.nabble.com/copy-of-Exchange-during-Multicast-tp5714477p5714528.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Re: copy of Exchange during Multicast

[Henryk Konsek <he...@gmail.com>]

> And each of the aggregated messages
> will got another header set on it (HEADER1 and HEADER2 respectively).

I meant KEY1 and KEY2 :) .

-- 
Henryk Konsek
http://henryk-konsek.blogspot.com

Re: copy of Exchange during Multicast

[Henryk Konsek <he...@gmail.com>]

Hi,

> Is there a way to make changes done to the
> header visible in further processing?

Yes. Actually modifications to the headers of the exchanged copied by
multicast are available in the aggregator.

In the example below GroupedExchangeAggregationStrategy will aggregate
two exchanges copied by multicast. And each of the aggregated messages
will got another header set on it (HEADER1 and HEADER2 respectively).

____________________________
CamelContext camelContext = new DefaultCamelContext();
camelContext.addRoutes(new RouteBuilder() {
  @Override
  public void configure() throws Exception {
    from("direct:start").multicast(new GroupedExchangeAggregationStrategy()).
      to("direct:headerModifier1", "direct:headerModifier2").
    end().
    to("log:AGGREGATED");

    from("direct:headerModifier1").process(new Processor() {
      @Override
      public void process(Exchange exchange) throws Exception {
        exchange.getIn().setHeader("KEY1", "VALUE1");
      }
    });

    from("direct:headerModifier2").process(new Processor() {
      @Override
      public void process(Exchange exchange) throws Exception {
        exchange.getIn().setHeader("KEY2", "VALUE2");
      }
    });
  }
});

camelContext.start();
camelContext.createProducerTemplate().sendBody("direct:start", "TestMe!");
____________________________


Kind regards.

-- 
Henryk Konsek
http://henryk-konsek.blogspot.com