content: /*
   *
   *  * Licensed to the Apache Software Foundation (ASF) under one or more
   *  * contributor license agreements.  See the NOTICE file distributed with
   *  * this work for additional information regarding copyright ownership.
   *  * The ASF licenses this file to You under the Apache License, Version 2.0
   *  * (the "License"); you may not use this file except in compliance with
   *  * the License.  You may obtain a copy of the License at
   *  *
   *  *      http://www.apache.org/licenses/LICENSE-2.0
   *  *
   *  * Unless required by applicable law or agreed to in writing, software
   *  * distributed under the License is distributed on an "AS IS" BASIS,
   *  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   *  * See the License for the specific language governing permissions and
   *  * limitations under the License.
   *
   */
  
  import java.util.ArrayList;
  import java.util.Arrays;
  
  import org.apache.camel.CamelContext;
  import org.apache.camel.CamelExchangeException;
  import org.apache.camel.CamelExecutionException;
  import org.apache.camel.ProducerTemplate;
  import org.apache.camel.component.mock.MockEndpoint;
  import org.apache.camel.example.spring.boot.Application;
  import org.apache.camel.test.spring.junit5.CamelSpringBootTest;
  import org.junit.jupiter.api.Test;
  
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.boot.test.context.SpringBootTest;
  
  import static org.apache.camel.test.junit5.TestSupport.assertIsInstanceOf;
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertThrows;
  
  @CamelSpringBootTest
  @SpringBootTest(classes = {Application.class})
  public class CamelSplitterEIPTests {
  
  	@Autowired
  	private CamelContext camelContext;
  
  	@Autowired
  	private ProducerTemplate producerTemplate;
  
  	@Test
  	public void testSplit() throws InterruptedException {
  		MockEndpoint mockA = camelContext.getEndpoint("mock:a", MockEndpoint.class);
  
  		mockA.expectedBodiesReceived("A", "B", "C");
  
  		producerTemplate.sendBody("direct:splitter", new ArrayList<>(Arrays.asList("A", "B", "C")));
  
  		mockA.assertIsSatisfied();
  	}
  
  	@Test
  	public void testSplitAggregate() throws InterruptedException {
  		MockEndpoint mockB = camelContext.getEndpoint("mock:b", MockEndpoint.class);
  		MockEndpoint mockC = camelContext.getEndpoint("mock:c", MockEndpoint.class);
  
  		mockB.expectedBodiesReceived("Alpha", "Beta", "Charlie");
  		mockC.expectedBodiesReceived("Alpha+Beta+Charlie");
  
  		producerTemplate.sendBody("direct:split-aggregate", "A,B,C");
  
  		mockB.assertIsSatisfied();
  		mockC.assertIsSatisfied();
  	}
  
  	@Test
  	public void testSplitAggregatePojo() throws InterruptedException {
  		MockEndpoint mockD = camelContext.getEndpoint("mock:d", MockEndpoint.class);
  		MockEndpoint mockE = camelContext.getEndpoint("mock:e", MockEndpoint.class);
  
  		mockD.expectedBodiesReceived("Alpha", "Beta", "Charlie");
  		mockE.expectedBodiesReceived("Alpha+Beta+Charlie");
  
  		producerTemplate.sendBody("direct:split-aggregate-bean", "A,B,C");
  
  		mockD.assertIsSatisfied();
  		mockE.assertIsSatisfied();
  	}
  
  	@Test
  	public void testSplitAggregateStopOnException() throws InterruptedException {
  		MockEndpoint mockF = camelContext.getEndpoint("mock:f", MockEndpoint.class);
  		MockEndpoint mockG = camelContext.getEndpoint("mock:g", MockEndpoint.class);
  
  		mockF.expectedBodiesReceived("Alpha");
  		mockG.expectedMessageCount(0);
  
  		final CamelExecutionException ex = assertThrows(CamelExecutionException.class,
  				() -> producerTemplate.sendBody("direct:split-aggregate-stop-on-exception", "A,E,C"));
  
  		CamelExchangeException cee = assertIsInstanceOf(CamelExchangeException.class, ex.getCause());
  		IllegalArgumentException iae = assertIsInstanceOf(IllegalArgumentException.class, cee.getCause());
  		assertEquals("Unknown key E", ex.getCause().getCause().getMessage());
  
  		mockF.assertIsSatisfied();
  		mockG.assertIsSatisfied();
  	}
  
  	@Test
  	public void testSplitAggregateStopOnAggregationException() throws InterruptedException {
  		MockEndpoint mockH = camelContext.getEndpoint("mock:h", MockEndpoint.class);
  		MockEndpoint mockJ = camelContext.getEndpoint("mock:j", MockEndpoint.class);
  
  		// without error aggregation is done as usual
  		mockH.expectedBodiesReceived("A", "B", "C");
  		mockJ.expectedBodiesReceived("A+B+C");
  
  		producerTemplate.sendBody("direct:split-aggregate-stop-on-aggregation-exception", "A,B,C");
  
  		mockH.assertIsSatisfied();
  		mockJ.assertIsSatisfied();
  
  		mockH.reset();
  		mockJ.reset();
  
  		// Received all messages that went into aggregation, including corrupted one
  		mockH.expectedBodiesReceived("A", "B", "E");
  		// Since aggregation stopped, receiving all original messages
  		mockJ.expectedBodiesReceived("A,B,E,C,D");
  
  		// 'E' is poison message which will lead to an error during aggregation
  		producerTemplate.sendBody("direct:split-aggregate-stop-on-aggregation-exception", "A,B,E,C,D");
  
  		mockH.assertIsSatisfied();
  		mockJ.assertIsSatisfied();
  	}
  }
  