I modified my route to look like below:
from("file:C:\\Users\\joe\\Desktop\\csv?noop=true").setHeader("isBuToPf",
simple("true")).
unmarshal().csv().process(new
RefDataProcessor()).to("direct:Process");
from("file:C:\\Users\\joe\\Desktop\\csv2?noop=true").setHeader("isBuToPf",
simple("true")).
unmarshal().csv().process(new
RefDataProcessor()).to("direct:Process");
from("direct:Process").aggregate(header("isBuToPf"), new
RefDataAggregationStrategy()).completionTimeout(3000).to("direct:Processed");
Now I get a NullPointerException in my aggregator. My aggregator looks like:
@Override
public Exchange aggregate(Exchange exchange, Exchange exchange2) {
final StringBuilder sb = new StringBuilder();
final Map < String, List < String > > buToPf =
exchange.getIn().getBody(Map.class); // NullPointer here...
final Map < String, List < String > > pfToUser =
exchange2.getIn().getBody(Map.class);
for (Map.Entry < String, List < String > > entry :
buToPf.entrySet()) {
for (String pf : entry.getValue()) {
List < String > userList = pfToUser.get(pf);
for (String user : userList) {
sb.append(entry.getKey() + "," + pf + "," + user +
"GO");
}
}
}
System.out.println(sb);
exchange.getIn().setBody(sb);
return exchange;
}
What I do in my RefDataProcessor.java is the following:
@Override
public void process(Exchange exchange) throws Exception {
final Map < String, List < String > > buToPfMap = new HashMap <
String, List < String > > ();
final List < List < String > > data = (List < List < String > > )
exchange.getIn().getBody();
for (List < String > line : data) {
putToMap(line.get(0), line.get(1), buToPfMap);
}
System.out.println(buToPfMap);
exchange.getIn().setBody(buToPfMap);
}
Why is there nothing in exchange?
--
View this message in context:
http://camel.465427.n5.nabble.com/Camel-Aggregator-Strategy-Question-tp5729337p5729555.html
Sent from the Camel - Users mailing list archive at Nabble.com.
