Test_oncompletion_agregate.java

package biz.daich.learning.camel;

import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.processor.aggregate.GroupedBodyAggregationStrategy;
import org.apache.camel.test.junit4.CamelTestSupport;
import org.junit.Test;

public class Test_oncompletion_agregate extends CamelTestSupport {

	@Override
	protected RouteBuilder createRouteBuilder() throws Exception {
		return new TestRoute();
	}

	private static class TestRoute extends RouteBuilder {
		@Override
		public void configure() throws Exception {

			// @formatter:off
			
			from("direct:start1")
					.enrich("direct:enrich")
					.end() // enrich
					.split()
						.tokenize(",")
						.streaming()
						.log("splited ${in.body}")
						.aggregate(constant(true), new GroupedBodyAggregationStrategy())
							.completionSize(2)
							.completionPredicate(header("CamelSplitComplete").isEqualTo(Boolean.TRUE)) // aggregate is done when Split is done see http://camel.apache.org/splitter.html for CamelSplitComplete  
							.log("in agregate ${in.body} and header CamelSplitComplete = ${header.CamelSplitComplete}")
							.to("direct:agregateSink") // here you can do what ever you want to do with the aggregated messages
						.end() // end of aggregate
					.end() // end of split
					.log("=== ROUTE 1 ==== DONE")
					.to("direct:start3")
					.to("mock:end1")
					.stop();
			
		
			// @formatter:on

			from("direct:start3")
					.log("=========ROUTE 3 started with <${in.body}>")
					.to("mock:end3");

			from("direct:enrich")
					.log("enrich requested with <${in.body}>")
					.transform().constant(testBody)
					.to("mock:enrich");

			from("direct:agregateSink")
					.log("GOT agregated exchanges: <${in.body}>")
					.to("mock:agregateSink");
		}
	}

	@Test
	public void test_0() throws Exception {
		log.info("Test started");

		// now set the expectations
		MockEndpoint end1 = getMockEndpoint("mock:end1");
		MockEndpoint end3 = getMockEndpoint("mock:end3");
		MockEndpoint agregateSink = getMockEndpoint("mock:agregateSink");

		assertNotNull(end1);
		assertNotNull(end3);
		assertNotNull(agregateSink);

		end1.expectedMessageCount(1);
		end3.expectedMessageCount(1);
		agregateSink.expectedMessageCount(5);

		// run the context
		context.start();

		// poke the route 1
		template.sendBody("direct:start1", "");

		// check that all expectations on the mocks are as expected
		assertMockEndpointsSatisfied();
	}

	public static final String testBody = "aaa,bbb,ccc,ddd,eee,fff,ggg,hhh,fff";
}