| /** |
| * 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. |
| */ |
| package org.apache.camel.builder; |
| |
| import java.util.ArrayList; |
| import java.util.Iterator; |
| import java.util.List; |
| |
| import org.apache.camel.CamelContext; |
| import org.apache.camel.Channel; |
| import org.apache.camel.Endpoint; |
| import org.apache.camel.Exchange; |
| import org.apache.camel.Processor; |
| import org.apache.camel.Producer; |
| import org.apache.camel.Route; |
| import org.apache.camel.TestSupport; |
| import org.apache.camel.impl.DefaultCamelContext; |
| import org.apache.camel.impl.EventDrivenConsumerRoute; |
| import org.apache.camel.processor.ChoiceProcessor; |
| import org.apache.camel.processor.DeadLetterChannel; |
| import org.apache.camel.processor.DelegateProcessor; |
| import org.apache.camel.processor.EvaluateExpressionProcessor; |
| import org.apache.camel.processor.FilterProcessor; |
| import org.apache.camel.processor.MulticastProcessor; |
| import org.apache.camel.processor.Pipeline; |
| import org.apache.camel.processor.RecipientList; |
| import org.apache.camel.processor.SendProcessor; |
| import org.apache.camel.processor.Splitter; |
| import org.apache.camel.processor.idempotent.IdempotentConsumer; |
| import org.apache.camel.processor.idempotent.MemoryIdempotentRepository; |
| |
| /** |
| * @version |
| */ |
| public class RouteBuilderTest extends TestSupport { |
| protected Processor myProcessor = new MyProcessor(); |
| protected DelegateProcessor interceptor1; |
| protected DelegateProcessor interceptor2; |
| |
| protected CamelContext createCamelContext() { |
| // disable stream cache otherwise to much hazzle in this unit test to filter the stream cache |
| // in all the assertion codes |
| DefaultCamelContext ctx = new DefaultCamelContext(); |
| ctx.setStreamCaching(Boolean.FALSE); |
| return ctx; |
| } |
| |
| protected List<Route> buildSimpleRoute() throws Exception { |
| // START SNIPPET: e1 |
| RouteBuilder builder = new RouteBuilder() { |
| public void configure() { |
| errorHandler(deadLetterChannel("mock:error")); |
| |
| from("seda:a").to("seda:b"); |
| } |
| }; |
| // END SNIPPET: e1 |
| return getRouteList(builder); |
| } |
| |
| public void testSimpleRoute() throws Exception { |
| List<Route> routes = buildSimpleRoute(); |
| |
| assertEquals("Number routes created", 1, routes.size()); |
| for (Route route : routes) { |
| Endpoint key = route.getEndpoint(); |
| assertEquals("From endpoint", "seda://a", key.getEndpointUri()); |
| |
| EventDrivenConsumerRoute consumer = assertIsInstanceOf(EventDrivenConsumerRoute.class, route); |
| Channel channel = unwrapChannel(consumer.getProcessor()); |
| |
| SendProcessor sendProcessor = assertIsInstanceOf(SendProcessor.class, channel.getNextProcessor()); |
| assertEquals("Endpoint URI", "seda://b", sendProcessor.getDestination().getEndpointUri()); |
| } |
| } |
| |
| protected List<Route> buildSimpleRouteWithHeaderPredicate() throws Exception { |
| // START SNIPPET: e2 |
| RouteBuilder builder = new RouteBuilder() { |
| public void configure() { |
| errorHandler(deadLetterChannel("mock:error")); |
| |
| from("seda:a") |
| .filter(header("foo").isEqualTo("bar")) |
| .to("seda:b"); |
| } |
| }; |
| // END SNIPPET: e2 |
| return getRouteList(builder); |
| } |
| |
| public void testSimpleRouteWithHeaderPredicate() throws Exception { |
| List<Route> routes = buildSimpleRouteWithHeaderPredicate(); |
| |
| log.debug("Created routes: " + routes); |
| |
| assertEquals("Number routes created", 1, routes.size()); |
| for (Route route : routes) { |
| Endpoint key = route.getEndpoint(); |
| assertEquals("From endpoint", "seda://a", key.getEndpointUri()); |
| |
| EventDrivenConsumerRoute consumer = assertIsInstanceOf(EventDrivenConsumerRoute.class, route); |
| Channel channel = unwrapChannel(consumer.getProcessor()); |
| |
| FilterProcessor filterProcessor = assertIsInstanceOf(FilterProcessor.class, channel.getNextProcessor()); |
| SendProcessor sendProcessor = assertIsInstanceOf(SendProcessor.class, unwrapChannel(filterProcessor).getNextProcessor()); |
| assertEquals("Endpoint URI", "seda://b", sendProcessor.getDestination().getEndpointUri()); |
| } |
| } |
| |
| protected List<Route> buildSimpleRouteWithChoice() throws Exception { |
| // START SNIPPET: e3 |
| RouteBuilder builder = new RouteBuilder() { |
| public void configure() { |
| errorHandler(deadLetterChannel("mock:error")); |
| |
| from("seda:a") |
| .choice() |
| .when(header("foo").isEqualTo("bar")) |
| .to("seda:b") |
| .when(header("foo").isEqualTo("cheese")) |
| .to("seda:c") |
| .otherwise() |
| .to("seda:d"); |
| } |
| }; |
| // END SNIPPET: e3 |
| return getRouteList(builder); |
| } |
| |
| public void testSimpleRouteWithChoice() throws Exception { |
| List<Route> routes = buildSimpleRouteWithChoice(); |
| |
| log.debug("Created routes: " + routes); |
| |
| assertEquals("Number routes created", 1, routes.size()); |
| for (Route route : routes) { |
| Endpoint key = route.getEndpoint(); |
| assertEquals("From endpoint", "seda://a", key.getEndpointUri()); |
| |
| EventDrivenConsumerRoute consumer = assertIsInstanceOf(EventDrivenConsumerRoute.class, route); |
| Channel channel = unwrapChannel(consumer.getProcessor()); |
| |
| ChoiceProcessor choiceProcessor = assertIsInstanceOf(ChoiceProcessor.class, channel.getNextProcessor()); |
| List<FilterProcessor> filters = choiceProcessor.getFilters(); |
| assertEquals("Should be two when clauses", 2, filters.size()); |
| |
| FilterProcessor filter1 = filters.get(0); |
| assertSendTo(unwrapChannel(filter1.getProcessor()).getNextProcessor(), "seda://b"); |
| |
| FilterProcessor filter2 = filters.get(1); |
| assertSendTo(unwrapChannel(filter2.getProcessor()).getNextProcessor(), "seda://c"); |
| |
| assertSendTo(unwrapChannel(choiceProcessor.getOtherwise()).getNextProcessor(), "seda://d"); |
| } |
| } |
| |
| protected List<Route> buildCustomProcessor() throws Exception { |
| // START SNIPPET: e4 |
| myProcessor = new Processor() { |
| public void process(Exchange exchange) { |
| log.debug("Called with exchange: " + exchange); |
| } |
| }; |
| |
| RouteBuilder builder = new RouteBuilder() { |
| public void configure() { |
| errorHandler(deadLetterChannel("mock:error")); |
| |
| from("seda:a") |
| .process(myProcessor); |
| } |
| }; |
| // END SNIPPET: e4 |
| return getRouteList(builder); |
| } |
| |
| public void testCustomProcessor() throws Exception { |
| List<Route> routes = buildCustomProcessor(); |
| |
| assertEquals("Number routes created", 1, routes.size()); |
| for (Route route : routes) { |
| Endpoint key = route.getEndpoint(); |
| assertEquals("From endpoint", "seda://a", key.getEndpointUri()); |
| } |
| } |
| |
| protected List<Route> buildCustomProcessorWithFilter() throws Exception { |
| // START SNIPPET: e5 |
| RouteBuilder builder = new RouteBuilder() { |
| public void configure() { |
| errorHandler(deadLetterChannel("mock:error")); |
| |
| from("seda:a") |
| .filter(header("foo").isEqualTo("bar")) |
| .process(myProcessor); |
| } |
| }; |
| // END SNIPPET: e5 |
| return getRouteList(builder); |
| } |
| |
| public void testCustomProcessorWithFilter() throws Exception { |
| List<Route> routes = buildCustomProcessorWithFilter(); |
| |
| log.debug("Created routes: " + routes); |
| |
| assertEquals("Number routes created", 1, routes.size()); |
| for (Route route : routes) { |
| Endpoint key = route.getEndpoint(); |
| assertEquals("From endpoint", "seda://a", key.getEndpointUri()); |
| } |
| } |
| |
| protected List<Route> buildWireTap() throws Exception { |
| // START SNIPPET: e6 |
| RouteBuilder builder = new RouteBuilder() { |
| public void configure() { |
| errorHandler(deadLetterChannel("mock:error")); |
| |
| from("seda:a") |
| .multicast().to("seda:tap", "seda:b"); |
| } |
| }; |
| // END SNIPPET: e6 |
| return getRouteList(builder); |
| } |
| |
| public void testWireTap() throws Exception { |
| List<Route> routes = buildWireTap(); |
| |
| log.debug("Created routes: " + routes); |
| |
| assertEquals("Number routes created", 1, routes.size()); |
| for (Route route : routes) { |
| Endpoint key = route.getEndpoint(); |
| assertEquals("From endpoint", "seda://a", key.getEndpointUri()); |
| |
| EventDrivenConsumerRoute consumer = assertIsInstanceOf(EventDrivenConsumerRoute.class, route); |
| Channel channel = unwrapChannel(consumer.getProcessor()); |
| |
| MulticastProcessor multicastProcessor = assertIsInstanceOf(MulticastProcessor.class, channel.getNextProcessor()); |
| List<Processor> endpoints = new ArrayList<Processor>(multicastProcessor.getProcessors()); |
| assertEquals("Should have 2 endpoints", 2, endpoints.size()); |
| |
| assertSendToProcessor(unwrapChannel(endpoints.get(0)).getNextProcessor(), "seda://tap"); |
| assertSendToProcessor(unwrapChannel(endpoints.get(1)).getNextProcessor(), "seda://b"); |
| } |
| } |
| |
| protected List<Route> buildRouteWithInterceptor() throws Exception { |
| interceptor1 = new DelegateProcessor() { |
| }; |
| |
| interceptor2 = new MyInterceptorProcessor(); |
| |
| RouteBuilder builder = new RouteBuilder() { |
| public void configure() { |
| errorHandler(deadLetterChannel("mock:error")); |
| |
| from("seda:a") |
| .process(interceptor1) |
| .process(interceptor2) |
| .to("seda:d"); |
| } |
| }; |
| return getRouteList(builder); |
| } |
| |
| public void testRouteWithInterceptor() throws Exception { |
| |
| List<Route> routes = buildRouteWithInterceptor(); |
| |
| log.debug("Created routes: " + routes); |
| |
| assertEquals("Number routes created", 1, routes.size()); |
| for (Route route : routes) { |
| Endpoint key = route.getEndpoint(); |
| assertEquals("From endpoint", "seda://a", key.getEndpointUri()); |
| |
| EventDrivenConsumerRoute consumer = assertIsInstanceOf(EventDrivenConsumerRoute.class, route); |
| |
| Pipeline line = assertIsInstanceOf(Pipeline.class, unwrap(consumer.getProcessor())); |
| assertEquals(3, line.getProcessors().size()); |
| // last should be our seda |
| |
| List<Processor> processors = new ArrayList<Processor>(line.getProcessors()); |
| Processor sendTo = assertIsInstanceOf(SendProcessor.class, unwrapChannel(processors.get(2)).getNextProcessor()); |
| assertSendTo(sendTo, "seda://d"); |
| } |
| } |
| |
| public void testComplexExpressions() throws Exception { |
| // START SNIPPET: e7 |
| RouteBuilder builder = new RouteBuilder() { |
| public void configure() { |
| errorHandler(deadLetterChannel("mock:error")); |
| |
| from("seda:a") |
| .filter(header("foo").isEqualTo(123)) |
| .to("seda:b"); |
| } |
| }; |
| // END SNIPPET: e7 |
| |
| List<Route> routes = getRouteList(builder); |
| log.debug("Created routes: " + routes); |
| |
| assertEquals("Number routes created", 1, routes.size()); |
| for (Route route : routes) { |
| Endpoint key = route.getEndpoint(); |
| assertEquals("From endpoint", "seda://a", key.getEndpointUri()); |
| } |
| } |
| |
| protected List<Route> buildStaticRecipientList() throws Exception { |
| // START SNIPPET: multicast |
| RouteBuilder builder = new RouteBuilder() { |
| public void configure() { |
| errorHandler(deadLetterChannel("mock:error")); |
| |
| from("seda:a") |
| .multicast().to("seda:b", "seda:c", "seda:d"); |
| } |
| }; |
| // END SNIPPET: multicast |
| return getRouteList(builder); |
| } |
| |
| protected List<Route> buildDynamicRecipientList() throws Exception { |
| // START SNIPPET: e9 |
| RouteBuilder builder = new RouteBuilder() { |
| public void configure() { |
| errorHandler(deadLetterChannel("mock:error")); |
| |
| from("seda:a") |
| .recipientList(header("foo")); |
| } |
| }; |
| // END SNIPPET: e9 |
| return getRouteList(builder); |
| } |
| |
| public void testRouteDynamicReceipentList() throws Exception { |
| |
| List<Route> routes = buildDynamicRecipientList(); |
| |
| log.debug("Created routes: " + routes); |
| |
| assertEquals("Number routes created", 1, routes.size()); |
| for (Route route : routes) { |
| Endpoint key = route.getEndpoint(); |
| assertEquals("From endpoint", "seda://a", key.getEndpointUri()); |
| |
| EventDrivenConsumerRoute consumer = assertIsInstanceOf(EventDrivenConsumerRoute.class, route); |
| Channel channel = unwrapChannel(consumer.getProcessor()); |
| |
| Pipeline line = assertIsInstanceOf(Pipeline.class, channel.getNextProcessor()); |
| Iterator it = line.getProcessors().iterator(); |
| |
| // EvaluateExpressionProcessor should be wrapped in error handler |
| Object first = it.next(); |
| first = assertIsInstanceOf(DeadLetterChannel.class, first).getOutput(); |
| assertIsInstanceOf(EvaluateExpressionProcessor.class, first); |
| |
| // and the second should NOT be wrapped in error handler |
| Object second = it.next(); |
| assertIsInstanceOf(RecipientList.class, second); |
| } |
| } |
| |
| protected List<Route> buildSplitter() throws Exception { |
| // START SNIPPET: splitter |
| RouteBuilder builder = new RouteBuilder() { |
| public void configure() { |
| errorHandler(deadLetterChannel("mock:error")); |
| |
| from("seda:a") |
| .split(body(String.class).tokenize("\n")) |
| .to("seda:b"); |
| } |
| }; |
| // END SNIPPET: splitter |
| return getRouteList(builder); |
| } |
| |
| public void testSplitter() throws Exception { |
| |
| List<Route> routes = buildSplitter(); |
| |
| log.debug("Created routes: " + routes); |
| |
| assertEquals("Number routes created", 1, routes.size()); |
| for (Route route : routes) { |
| Endpoint key = route.getEndpoint(); |
| assertEquals("From endpoint", "seda://a", key.getEndpointUri()); |
| |
| EventDrivenConsumerRoute consumer = assertIsInstanceOf(EventDrivenConsumerRoute.class, route); |
| Channel channel = unwrapChannel(consumer.getProcessor()); |
| assertIsInstanceOf(Splitter.class, channel.getNextProcessor()); |
| } |
| } |
| |
| protected List<Route> buildIdempotentConsumer() throws Exception { |
| // START SNIPPET: idempotent |
| RouteBuilder builder = new RouteBuilder() { |
| public void configure() { |
| errorHandler(deadLetterChannel("mock:error")); |
| |
| from("seda:a") |
| .idempotentConsumer(header("myMessageId"), |
| MemoryIdempotentRepository.memoryIdempotentRepository(200)) |
| .to("seda:b"); |
| } |
| }; |
| // END SNIPPET: idempotent |
| return getRouteList(builder); |
| } |
| |
| public void testIdempotentConsumer() throws Exception { |
| |
| List<Route> routes = buildIdempotentConsumer(); |
| |
| log.debug("Created routes: " + routes); |
| |
| assertEquals("Number routes created", 1, routes.size()); |
| for (Route route : routes) { |
| Endpoint key = route.getEndpoint(); |
| assertEquals("From endpoint", "seda://a", key.getEndpointUri()); |
| |
| EventDrivenConsumerRoute consumer = assertIsInstanceOf(EventDrivenConsumerRoute.class, route); |
| Channel channel = unwrapChannel(consumer.getProcessor()); |
| |
| IdempotentConsumer idempotentConsumer = assertIsInstanceOf(IdempotentConsumer.class, channel.getNextProcessor()); |
| assertEquals("messageIdExpression", "header(myMessageId)", idempotentConsumer.getMessageIdExpression().toString()); |
| |
| assertIsInstanceOf(MemoryIdempotentRepository.class, idempotentConsumer.getIdempotentRepository()); |
| SendProcessor sendProcessor = assertIsInstanceOf(SendProcessor.class, unwrapChannel(idempotentConsumer.getProcessor()).getNextProcessor()); |
| assertEquals("Endpoint URI", "seda://b", sendProcessor.getDestination().getEndpointUri()); |
| } |
| } |
| |
| protected void assertSendTo(Processor processor, String uri) { |
| if (!(processor instanceof SendProcessor)) { |
| processor = unwrapErrorHandler(processor); |
| } |
| |
| SendProcessor sendProcessor = assertIsInstanceOf(SendProcessor.class, processor); |
| assertEquals("Endpoint URI", uri, sendProcessor.getDestination().getEndpointUri()); |
| } |
| |
| protected void assertSendToProcessor(Processor processor, String uri) { |
| if (!(processor instanceof Producer)) { |
| processor = unwrapErrorHandler(processor); |
| } |
| |
| if (processor instanceof SendProcessor) { |
| assertSendTo(processor, uri); |
| } else { |
| Producer producer = assertIsInstanceOf(Producer.class, processor); |
| assertEquals("Endpoint URI", uri, producer.getEndpoint().getEndpointUri()); |
| } |
| } |
| |
| /** |
| * By default routes should be wrapped in the {@link DeadLetterChannel} so |
| * lets unwrap that and return the actual processor |
| */ |
| protected Processor getProcessorWithoutErrorHandler(Route route) { |
| EventDrivenConsumerRoute consumerRoute = assertIsInstanceOf(EventDrivenConsumerRoute.class, route); |
| Processor processor = unwrap(consumerRoute.getProcessor()); |
| return unwrapErrorHandler(processor); |
| } |
| |
| protected Processor unwrapErrorHandler(Processor processor) { |
| if (processor instanceof DeadLetterChannel) { |
| DeadLetterChannel deadLetter = (DeadLetterChannel)processor; |
| return deadLetter.getOutput(); |
| } else { |
| return processor; |
| } |
| } |
| |
| protected Processor unwrapDelegateProcessor(Processor processor) { |
| if (processor instanceof DelegateProcessor) { |
| DelegateProcessor delegate = (DelegateProcessor) processor; |
| return delegate.getProcessor(); |
| } else { |
| return processor; |
| } |
| } |
| |
| public void testCorrectNumberOfRoutes() throws Exception { |
| RouteBuilder builder = new RouteBuilder() { |
| public void configure() throws Exception { |
| errorHandler(deadLetterChannel("mock:error")); |
| |
| from("direct:start").to("seda:in"); |
| |
| from("seda:in").to("mock:result"); |
| } |
| }; |
| |
| List<Route> routes = getRouteList(builder); |
| |
| assertEquals(2, routes.size()); |
| } |
| } |