camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject svn commit: r1327537 - in /camel/trunk/camel-core/src: main/java/org/apache/camel/impl/ test/java/org/apache/camel/component/file/ test/java/org/apache/camel/processor/
Date Wed, 18 Apr 2012 15:08:49 GMT
Author: davsclaus
Date: Wed Apr 18 15:08:48 2012
New Revision: 1327537

URL: http://svn.apache.org/viewvc?rev=1327537&view=rev
Log:
CAMEL-1260: Added support for consumer.bridgeErrorHandler option to let the routing error
handler deal with cosumer exceptions occuring outside the routing engine.

Added:
    camel/trunk/camel-core/src/main/java/org/apache/camel/impl/BridgeExceptionHandlerToErrorHandler.java
    camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerBridgeRouteExceptionHandlerTest.java
      - copied, changed from r1327499, camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerCustomExceptionHandlerTest.java
    camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DefaultConsumerBridgeErrorHandlerTest.java
Modified:
    camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultEndpoint.java
    camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ScheduledPollEndpoint.java

Added: camel/trunk/camel-core/src/main/java/org/apache/camel/impl/BridgeExceptionHandlerToErrorHandler.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/BridgeExceptionHandlerToErrorHandler.java?rev=1327537&view=auto
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/impl/BridgeExceptionHandlerToErrorHandler.java
(added)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/impl/BridgeExceptionHandlerToErrorHandler.java
Wed Apr 18 15:08:48 2012
@@ -0,0 +1,73 @@
+/**
+ * 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.impl;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.spi.ExceptionHandler;
+
+/**
+ * An {@link ExceptionHandler} that uses the {@link DefaultConsumer} to
+ * process the caused exception to send the message into the Camel routing engine
+ * which allows to let the routing engine handle the exception.
+ * <p/>
+ * An endpoint can be configured with <tt>consumer.bridgeErrorHandler=true</tt>
in the URI
+ * to enable this {@link BridgeExceptionHandlerToErrorHandler} on the consumer.
+ * The consumer must extend the {@link DefaultConsumer}, to support this, if not an
+ * {@link IllegalArgumentException} is thrown upon startup.
+ */
+public class BridgeExceptionHandlerToErrorHandler implements ExceptionHandler {
+
+    private final LoggingExceptionHandler fallback;
+    private final Consumer consumer;
+    private final Processor bridge;
+
+    public BridgeExceptionHandlerToErrorHandler(DefaultConsumer consumer) {
+        this.consumer = consumer;
+        this.fallback = new LoggingExceptionHandler(consumer.getClass());
+        this.bridge = consumer.getProcessor();
+    }
+
+    @Override
+    public void handleException(Throwable exception) {
+        handleException(null, exception);
+    }
+
+    @Override
+    public void handleException(String message, Throwable exception) {
+        handleException(message, null, exception);
+    }
+
+    @Override
+    public void handleException(String message, Exchange exchange, Throwable exception) {
+        if (exchange == null) {
+            exchange = consumer.getEndpoint().createExchange();
+        }
+
+        // set the caused exception
+        exchange.setException(exception);
+        // and the message
+        exchange.getIn().setBody(message);
+
+        try {
+            bridge.process(exchange);
+        } catch (Exception e) {
+            fallback.handleException("Error handling exception " + exception.getMessage(),
exchange, e);
+        }
+    }
+}

Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultEndpoint.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultEndpoint.java?rev=1327537&r1=1327536&r2=1327537&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultEndpoint.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultEndpoint.java Wed Apr
18 15:08:48 2012
@@ -19,21 +19,25 @@ package org.apache.camel.impl;
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
 import java.net.URI;
+import java.util.HashMap;
 import java.util.Map;
 
 import org.apache.camel.CamelContext;
 import org.apache.camel.CamelContextAware;
 import org.apache.camel.Component;
+import org.apache.camel.Consumer;
 import org.apache.camel.Endpoint;
 import org.apache.camel.EndpointConfiguration;
 import org.apache.camel.EndpointConfiguration.UriFormat;
 import org.apache.camel.Exchange;
 import org.apache.camel.ExchangePattern;
 import org.apache.camel.PollingConsumer;
+import org.apache.camel.ResolveEndpointFailedException;
 import org.apache.camel.RuntimeCamelException;
 import org.apache.camel.spi.HasId;
 import org.apache.camel.support.ServiceSupport;
 import org.apache.camel.util.EndpointHelper;
+import org.apache.camel.util.IntrospectionSupport;
 import org.apache.camel.util.ObjectHelper;
 import org.apache.camel.util.URISupport;
 
@@ -61,6 +65,7 @@ public abstract class DefaultEndpoint ex
     // used or not (if possible)
     private boolean synchronous;
     private final String id = EndpointHelper.createEndpointId();
+    private Map<String, Object> consumerProperties;
 
     /**
      * Constructs a fully-initialized DefaultEndpoint instance. This is the
@@ -272,7 +277,10 @@ public abstract class DefaultEndpoint ex
     }
 
     public void configureProperties(Map<String, Object> options) {
-        // do nothing by default
+        Map<String, Object> consumerProperties = IntrospectionSupport.extractProperties(options,
"consumer.");
+        if (consumerProperties != null) {
+            setConsumerProperties(consumerProperties);
+        }
     }
 
     /**
@@ -327,6 +335,45 @@ public abstract class DefaultEndpoint ex
         return false;
     }
 
+    public Map<String, Object> getConsumerProperties() {
+        return consumerProperties;
+    }
+
+    public void setConsumerProperties(Map<String, Object> consumerProperties) {
+        this.consumerProperties = consumerProperties;
+    }
+
+    protected void configureConsumer(Consumer consumer) throws Exception {
+        if (consumerProperties != null) {
+            // use a defensive copy of the consumer properties as the methods below will
remove the used properties
+            // and in case we restart routes, we need access to the original consumer properties
again
+            Map<String, Object> copy = new HashMap<String, Object>(consumerProperties);
+
+            // set reference properties first as they use # syntax that fools the regular
properties setter
+            EndpointHelper.setReferenceProperties(getCamelContext(), consumer, copy);
+            EndpointHelper.setProperties(getCamelContext(), consumer, copy);
+
+            // special consumer.routeExceptionHandler option
+            Object bridge = copy.remove("bridgeErrorHandler");
+            if (bridge != null && "true".equals(bridge)) {
+                if (consumer instanceof DefaultConsumer) {
+                    DefaultConsumer defaultConsumer = (DefaultConsumer) consumer;
+                    defaultConsumer.setExceptionHandler(new BridgeExceptionHandlerToErrorHandler(defaultConsumer));
+                } else {
+                    throw new IllegalArgumentException("Option consumer.bridgeErrorHandler
is only supported by endpoints," +
+                            "having the consumer extend the DefaultConsumer. The consumer
is a " + consumer.getClass().getName() + " class.");
+                }
+            }
+
+            if (!this.isLenientProperties() && copy.size() > 0) {
+                throw new ResolveEndpointFailedException(this.getEndpointUri(), "There are
" + copy.size()
+                    + " parameters that couldn't be set on the endpoint consumer."
+                    + " Check the uri if the parameters are spelt correctly and that they
are properties of the endpoint."
+                    + " Unknown consumer parameters=[" + copy + "]");
+            }
+        }
+    }
+
     @Override
     protected void doStart() throws Exception {
         // noop

Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ScheduledPollEndpoint.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ScheduledPollEndpoint.java?rev=1327537&r1=1327536&r2=1327537&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ScheduledPollEndpoint.java
(original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ScheduledPollEndpoint.java
Wed Apr 18 15:08:48 2012
@@ -21,10 +21,6 @@ import java.util.Map;
 
 import org.apache.camel.CamelContext;
 import org.apache.camel.Component;
-import org.apache.camel.Consumer;
-import org.apache.camel.ResolveEndpointFailedException;
-import org.apache.camel.util.EndpointHelper;
-import org.apache.camel.util.IntrospectionSupport;
 
 /**
  * A base class for {@link org.apache.camel.Endpoint} which creates a {@link ScheduledPollConsumer}
@@ -32,7 +28,6 @@ import org.apache.camel.util.Introspecti
  * @version 
  */
 public abstract class ScheduledPollEndpoint extends DefaultEndpoint {
-    private Map<String, Object> consumerProperties;
 
     protected ScheduledPollEndpoint(String endpointUri, Component component) {
         super(endpointUri, component);
@@ -51,38 +46,9 @@ public abstract class ScheduledPollEndpo
     protected ScheduledPollEndpoint() {
     }
 
-    public Map<String, Object> getConsumerProperties() {
-        return consumerProperties;
-    }
-
-    public void setConsumerProperties(Map<String, Object> consumerProperties) {
-        this.consumerProperties = consumerProperties;
-    }
-
-    protected void configureConsumer(Consumer consumer) throws Exception {
-        if (consumerProperties != null) {
-            // use a defensive copy of the consumer properties as the methods below will
remove the used properties
-            // and in case we restart routes, we need access to the original consumer properties
again
-            Map<String, Object> copy = new HashMap<String, Object>(consumerProperties);
-
-            // set reference properties first as they use # syntax that fools the regular
properties setter
-            EndpointHelper.setReferenceProperties(getCamelContext(), consumer, copy);
-            EndpointHelper.setProperties(getCamelContext(), consumer, copy);
-            if (!this.isLenientProperties() && copy.size() > 0) {
-                throw new ResolveEndpointFailedException(this.getEndpointUri(), "There are
" + copy.size()
-                    + " parameters that couldn't be set on the endpoint consumer."
-                    + " Check the uri if the parameters are spelt correctly and that they
are properties of the endpoint."
-                    + " Unknown consumer parameters=[" + copy + "]");
-            }
-        }
-    }
-
     public void configureProperties(Map<String, Object> options) {
-        Map<String, Object> consumerProperties = IntrospectionSupport.extractProperties(options,
"consumer.");
-        if (consumerProperties != null) {
-            setConsumerProperties(consumerProperties);
-        }
-        configureScheduledPollConsumerProperties(options, consumerProperties);
+        super.configureProperties(options);
+        configureScheduledPollConsumerProperties(options, getConsumerProperties());
     }
 
     private void configureScheduledPollConsumerProperties(Map<String, Object> options,
Map<String, Object> consumerProperties) {

Copied: camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerBridgeRouteExceptionHandlerTest.java
(from r1327499, camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerCustomExceptionHandlerTest.java)
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerBridgeRouteExceptionHandlerTest.java?p2=camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerBridgeRouteExceptionHandlerTest.java&p1=camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerCustomExceptionHandlerTest.java&r1=1327499&r2=1327537&rev=1327537&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerCustomExceptionHandlerTest.java
(original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerBridgeRouteExceptionHandlerTest.java
Wed Apr 18 15:08:48 2012
@@ -20,23 +20,17 @@ import java.io.IOException;
 
 import org.apache.camel.ContextTestSupport;
 import org.apache.camel.Exchange;
-import org.apache.camel.Processor;
-import org.apache.camel.ProducerTemplate;
 import org.apache.camel.builder.RouteBuilder;
 import org.apache.camel.impl.JndiRegistry;
-import org.apache.camel.spi.ExceptionHandler;
 
 /**
  *
  */
-public class FileConsumerCustomExceptionHandlerTest extends ContextTestSupport {
+public class FileConsumerBridgeRouteExceptionHandlerTest extends ContextTestSupport {
 
     private MyReadLockStrategy myReadLockStrategy = new MyReadLockStrategy();
-    private MyExceptionHandler myExceptionHandler = new MyExceptionHandler();
 
     public void testCustomExceptionHandler() throws Exception {
-        myExceptionHandler.setTemplate(context.createProducerTemplate());
-
         getMockEndpoint("mock:result").expectedMessageCount(2);
         getMockEndpoint("mock:error").expectedBodiesReceived("Error Forced to simulate no
space on device");
 
@@ -51,7 +45,6 @@ public class FileConsumerCustomException
     @Override
     protected JndiRegistry createRegistry() throws Exception {
         JndiRegistry jndi = super.createRegistry();
-        jndi.bind("myExceptionHandler", myExceptionHandler);
         jndi.bind("myReadLockStrategy", myReadLockStrategy);
         return jndi;
     }
@@ -66,20 +59,12 @@ public class FileConsumerCustomException
                 onException(IOException.class)
                     .handled(true)
                     .log("IOException occurred due: ${exception.message}")
-                    // as we handle the exception we can send it to direct:file-error,
-                    // where we could send out alerts or whatever we want
-                    .to("direct:file-error");
-
-                // special route that handles file errors
-                from("direct:file-error")
-                    .log("File error route triggered to deal with exception ${exception?.class}")
-                    // as this is based on unit test just transform a message and send it
to a mock
                     .transform().simple("Error ${exception.message}")
                     .to("mock:error");
 
-                // this is the file route that pickup files, notice how we use our custom
exception handler on the consumer
+                // this is the file route that pickup files, notice how we bridge the consumer
to use the Camel routing error handler
                 // the exclusiveReadLockStrategy is only configured because this is from
an unit test, so we use that to simulate exceptions
-                from("file:target/nospace?exclusiveReadLockStrategy=#myReadLockStrategy&consumer.exceptionHandler=#myExceptionHandler")
+                from("file:target/nospace?exclusiveReadLockStrategy=#myReadLockStrategy&consumer.bridgeErrorHandler=true")
                     .convertBodyTo(String.class)
                     .to("mock:result");
             }
@@ -87,48 +72,6 @@ public class FileConsumerCustomException
     }
     // END SNIPPET: e2
 
-    // START SNIPPET: e1
-    /**
-     * Custom {@link ExceptionHandler} to be used on the file consumer, to send
-     * exceptions to a Camel route, to let Camel deal with the error.
-     */
-    private class MyExceptionHandler implements ExceptionHandler {
-
-        private ProducerTemplate template;
-
-        /**
-         * We use a producer template to send a message to the Camel route
-         */
-        public void setTemplate(ProducerTemplate template) {
-            this.template = template;
-        }
-
-        @Override
-        public void handleException(Throwable exception) {
-            handleException(exception.getMessage(), exception);
-        }
-
-        @Override
-        public void handleException(String message, Throwable exception) {
-            handleException(exception.getMessage(), null, exception);
-        }
-
-        @Override
-        public void handleException(final String message, final Exchange originalExchange,
final Throwable exception) {
-            // send the message to the special direct:file-error endpoint, which will trigger
exception handling
-            //
-            template.send("direct:file-error", new Processor() {
-                @Override
-                public void process(Exchange exchange) throws Exception {
-                    // set an exception on the message from the start so the error handling
is triggered
-                    exchange.setException(exception);
-                    exchange.getIn().setBody(message);
-                }
-            });
-        }
-    }
-    // END SNIPPET: e1
-
     // used for simulating exception during acquiring a lock on the file
     private class MyReadLockStrategy implements GenericFileExclusiveReadLockStrategy {
 

Added: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DefaultConsumerBridgeErrorHandlerTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DefaultConsumerBridgeErrorHandlerTest.java?rev=1327537&view=auto
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DefaultConsumerBridgeErrorHandlerTest.java
(added)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DefaultConsumerBridgeErrorHandlerTest.java
Wed Apr 18 15:08:48 2012
@@ -0,0 +1,154 @@
+/**
+ * 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.processor;
+
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.Component;
+import org.apache.camel.Consumer;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.DefaultComponent;
+import org.apache.camel.impl.DefaultConsumer;
+import org.apache.camel.impl.DefaultEndpoint;
+
+/**
+ *
+ */
+public class DefaultConsumerBridgeErrorHandlerTest extends ContextTestSupport {
+
+    private final CountDownLatch latch = new CountDownLatch(1);
+
+    public void testDefaultConsumerBridgeErrorHandler() throws Exception {
+        getMockEndpoint("mock:result").expectedBodiesReceived("Hello World", "Hello World");
+        getMockEndpoint("mock:dead").expectedBodiesReceived("Cannot process");
+
+        latch.countDown();
+
+        assertMockEndpointsSatisfied();
+
+        Exception cause = getMockEndpoint("mock:dead").getReceivedExchanges().get(0).getProperty(Exchange.EXCEPTION_CAUGHT,
Exception.class);
+        assertNotNull(cause);
+        assertEquals("Simulated", cause.getMessage());
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() throws Exception {
+        // START SNIPPET: e1
+        return new RouteBuilder() {
+            @Override
+            public void configure() throws Exception {
+                // register our custom component
+                getContext().addComponent("my", new MyComponent());
+
+                // configure error handler
+                errorHandler(deadLetterChannel("mock:dead"));
+
+                // configure the consumer to bridge with the Camel error handler,
+                // so the above error handler will trigger if exceptions also
+                // occurs inside the consumer
+                from("my:foo?consumer.bridgeErrorHandler=true")
+                    .to("log:foo")
+                    .to("mock:result");
+            }
+        };
+        // END SNIPPET: e1
+    }
+
+    public class MyComponent extends DefaultComponent {
+
+        @Override
+        protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object>
parameters) throws Exception {
+            return new MyEndpoint(uri, this);
+        }
+    }
+
+    public class MyEndpoint extends DefaultEndpoint {
+
+        public MyEndpoint(String endpointUri, Component component) {
+            super(endpointUri, component);
+        }
+
+        @Override
+        public Producer createProducer() throws Exception {
+            return null;
+        }
+
+        @Override
+        public Consumer createConsumer(Processor processor) throws Exception {
+            Consumer answer = new MyConsumer(this, processor);
+            configureConsumer(answer);
+            return answer;
+        }
+
+        @Override
+        public boolean isSingleton() {
+            return true;
+        }
+    }
+
+    public class MyConsumer extends DefaultConsumer {
+
+        private int invoked;
+
+        public MyConsumer(Endpoint endpoint, Processor processor) {
+            super(endpoint, processor);
+        }
+
+        public void doSomething() throws Exception {
+            try {
+                if (invoked++ == 0) {
+                    throw new IllegalArgumentException("Simulated");
+                }
+
+                Exchange exchange = getEndpoint().createExchange();
+                exchange.getIn().setBody("Hello World");
+                getProcessor().process(exchange);
+
+            } catch (Exception e) {
+                getExceptionHandler().handleException("Cannot process", e);
+            }
+        }
+
+        @Override
+        protected void doStart() throws Exception {
+            super.doStart();
+
+            Thread thread = new Thread() {
+                @Override
+                public void run() {
+                    try {
+                        // do not start before the mocks has been setup and is ready
+                        latch.await(5, TimeUnit.SECONDS);
+                        doSomething();
+                        doSomething();
+                        doSomething();
+                    } catch (Exception e) {
+                        // ignore
+                    }
+                }
+            };
+            thread.start();
+        }
+    }
+}



Mime
View raw message