camel.java

public class Principal {
	public static void main(String[] args) throws Exception {
		DefaultCamelContext context = new DefaultCamelContext();
		
		context.addRoutes(new RouteBuilder() {
			
			AggregationStrategy ag = new MyAggragate(); 
			
			@Override
			public void configure() throws Exception {
				from("file:entrada?delay=1s&noop=true")
				.split()
					.xpath("/pedido/itens/item/formato[text()='EBOOK']/..")
				.to("direct:aggregate-item");				
				
				from("direct:aggregate-item")
					.aggregate(constant(true), ag)
						.completionSize(header("CamelSplitSize"))
						.transform(simple("<itens>\n${body}\n</itens>"))
						.log("${body}")
						.log("${headers.CamelFileAbsolutePath}")
						.setBody(simple("insert into Livros values('${headers.CamelFileAbsolutePath}')"))
						.log("${body}")
						.to("mock:jms/FILA.GERADOR");
						
			}
		});
		
		context.start();
		Thread.sleep(2000);
		context.stop();
	}
	private static class MyAggragate implements AggregationStrategy{

		@Override
		public Exchange aggregate(Exchange a, Exchange b) {
			
			if (a == null) return b;
			
			String aBody = a.getIn().getBody(String.class);
			String bBody = b.getIn().getBody(String.class);
			
			a.getIn().setBody(aBody + "\n" + bBody);
						
			return a;
		} 
		
	}
}