Repository: camel Updated Branches: refs/heads/master c9e1bf33a -> 7e9047af4
CAMEL-10643: camel-sjms - Should log caught JMS exceptions by default. Project: http://git-wip-us.apache.org/repos/asf/camel/repo Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/7e9047af Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/7e9047af Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/7e9047af Branch: refs/heads/master Commit: 7e9047af4df40d7664957bf3bb942e9ae9c87c20 Parents: c9e1bf3 Author: Claus Ibsen <[email protected]> Authored: Thu Dec 22 13:19:02 2016 +0100 Committer: Claus Ibsen <[email protected]> Committed: Thu Dec 22 13:26:36 2016 +0100 ---------------------------------------------------------------------- .../src/main/docs/sjms-component.adoc | 5 +- .../camel/component/sjms/SjmsConsumer.java | 3 +- .../camel/component/sjms/SjmsEndpoint.java | 54 +++++++++++++++- .../sjms/SjmsLoggingExceptionListener.java | 45 +++++++++++++ .../camel/component/sjms/SjmsProducer.java | 2 +- .../sjms/jms/ConnectionFactoryResource.java | 15 ++++- .../sjms/manual/ManualFromQueueTest.java | 68 ++++++++++++++++++++ 7 files changed, 186 insertions(+), 6 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/camel/blob/7e9047af/components/camel-sjms/src/main/docs/sjms-component.adoc ---------------------------------------------------------------------- diff --git a/components/camel-sjms/src/main/docs/sjms-component.adoc b/components/camel-sjms/src/main/docs/sjms-component.adoc index 34c3fff..1132a49 100644 --- a/components/camel-sjms/src/main/docs/sjms-component.adoc +++ b/components/camel-sjms/src/main/docs/sjms-component.adoc @@ -133,7 +133,7 @@ The Simple JMS component supports 9 options which are listed below. // endpoint options: START -The Simple JMS component supports 33 endpoint options which are listed below: +The Simple JMS component supports 36 endpoint options which are listed below: {% raw %} [width="100%",cols="2,1,1m,1m,5",options="header"] @@ -162,6 +162,9 @@ The Simple JMS component supports 33 endpoint options which are listed below: | connectionFactory | advanced | | ConnectionFactory | Initializes the connectionFactory for the endpoint which takes precedence over the component's connectionFactory if any | connectionResource | advanced | | ConnectionResource | Initializes the connectionResource for the endpoint which takes precedence over the component's connectionResource if any | destinationCreationStrategy | advanced | | DestinationCreationStrategy | To use a custom DestinationCreationStrategy. +| errorHandlerLoggingLevel | advanced | WARN | LoggingLevel | Allows to configure the default errorHandler logging level for logging uncaught exceptions. +| errorHandlerLogStackTrace | advanced | true | boolean | Allows to control whether stacktraces should be logged or not by the default errorHandler. +| exceptionListener | advanced | | ExceptionListener | Specifies the JMS Exception Listener that is to be notified of any underlying JMS exceptions. | headerFilterStrategy | advanced | | HeaderFilterStrategy | To use a custom HeaderFilterStrategy to filter header to and from Camel message. | includeAllJMSXProperties | advanced | false | boolean | Whether to include all JMSXxxx properties when mapping from JMS to Camel Message. Setting this to true will include properties such as JMSXAppID and JMSXUserID etc. Note: If you are using a custom headerFilterStrategy then this option does not apply. | jmsKeyFormatStrategy | advanced | | JmsKeyFormatStrategy | Pluggable strategy for encoding and decoding JMS keys so they can be compliant with the JMS specification. Camel provides two implementations out of the box: default and passthrough. The default strategy will safely marshal dots and hyphens (. and -). The passthrough strategy leaves the key as is. Can be used for JMS brokers which do not care whether JMS header keys contain illegal characters. You can provide your own implementation of the org.apache.camel.component.jms.JmsKeyFormatStrategy and refer to it using the notation. http://git-wip-us.apache.org/repos/asf/camel/blob/7e9047af/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsConsumer.java ---------------------------------------------------------------------- diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsConsumer.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsConsumer.java index f6064c2..1345b91 100644 --- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsConsumer.java +++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsConsumer.java @@ -239,6 +239,7 @@ public class SjmsConsumer extends DefaultConsumer { messageHandler = new InOutMessageHandler(getEndpoint(), executor); } } + messageHandler.setSession(session); messageHandler.setProcessor(getAsyncProcessor()); messageHandler.setSynchronous(isSynchronous()); @@ -259,7 +260,7 @@ public class SjmsConsumer extends DefaultConsumer { protected ConnectionResource getOrCreateConnectionResource() { ConnectionResource answer = getEndpoint().getConnectionResource(); if (answer == null) { - answer = getEndpoint().createConnectionResource(); + answer = getEndpoint().createConnectionResource(this); } return answer; } http://git-wip-us.apache.org/repos/asf/camel/blob/7e9047af/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java ---------------------------------------------------------------------- diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java index 35a2d89..16e9573 100644 --- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java +++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java @@ -17,6 +17,7 @@ package org.apache.camel.component.sjms; import javax.jms.ConnectionFactory; +import javax.jms.ExceptionListener; import javax.jms.Message; import javax.jms.Session; @@ -25,6 +26,7 @@ import org.apache.camel.Component; import org.apache.camel.Consumer; import org.apache.camel.Exchange; import org.apache.camel.ExchangePattern; +import org.apache.camel.LoggingLevel; import org.apache.camel.MultipleConsumersSupport; import org.apache.camel.Processor; import org.apache.camel.Producer; @@ -47,6 +49,7 @@ import org.apache.camel.spi.Metadata; import org.apache.camel.spi.UriEndpoint; import org.apache.camel.spi.UriParam; import org.apache.camel.spi.UriPath; +import org.apache.camel.support.LoggingExceptionHandler; import org.apache.camel.util.EndpointHelper; import org.apache.camel.util.ObjectHelper; import org.slf4j.Logger; @@ -127,6 +130,13 @@ public class SjmsEndpoint extends DefaultEndpoint implements AsyncEndpoint, Mult private ConnectionFactory connectionFactory; @UriParam(label = "advanced") private Integer connectionCount; + @UriParam(label = "advanced") + private ExceptionListener exceptionListener; + @UriParam(defaultValue = "WARN", label = "advanced") + private LoggingLevel errorHandlerLoggingLevel = LoggingLevel.WARN; + @UriParam(defaultValue = "true", label = "advanced") + private boolean errorHandlerLogStackTrace = true; + private volatile boolean closeConnectionResource; public SjmsEndpoint() { @@ -152,7 +162,7 @@ public class SjmsEndpoint extends DefaultEndpoint implements AsyncEndpoint, Mult // if we are not async starting then create connection eager if (getConnectionResource() == null) { if (getConnectionFactory() != null) { - connectionResource = createConnectionResource(); + connectionResource = createConnectionResource(this); // we created the resource so we should close it when stopping closeConnectionResource = true; } @@ -202,7 +212,7 @@ public class SjmsEndpoint extends DefaultEndpoint implements AsyncEndpoint, Mult return true; } - protected ConnectionResource createConnectionResource() { + protected ConnectionResource createConnectionResource(Object source) { if (getConnectionFactory() == null) { throw new IllegalArgumentException(String.format("ConnectionResource or ConnectionFactory must be configured for %s", this)); } @@ -211,6 +221,13 @@ public class SjmsEndpoint extends DefaultEndpoint implements AsyncEndpoint, Mult logger.debug("Creating ConnectionResource with connectionCount: {} using ConnectionFactory", getConnectionCount(), getConnectionFactory()); // We always use a connection pool, even for a pool of 1 ConnectionFactoryResource connections = new ConnectionFactoryResource(getConnectionCount(), getConnectionFactory()); + if (exceptionListener != null) { + connections.setExceptionListener(exceptionListener); + } else { + // add a exception listener that logs so we can see any errors that happens + ExceptionListener listener = new SjmsLoggingExceptionListener(new LoggingExceptionHandler(getCamelContext(), source.getClass()), isErrorHandlerLogStackTrace()); + connections.setExceptionListener(listener); + } connections.fillPool(); return connections; } catch (Exception e) { @@ -637,4 +654,37 @@ public class SjmsEndpoint extends DefaultEndpoint implements AsyncEndpoint, Mult public void setConnectionCount(Integer connectionCount) { this.connectionCount = connectionCount; } + + public ExceptionListener getExceptionListener() { + return exceptionListener; + } + + /** + * Specifies the JMS Exception Listener that is to be notified of any underlying JMS exceptions. + */ + public void setExceptionListener(ExceptionListener exceptionListener) { + this.exceptionListener = exceptionListener; + } + + public LoggingLevel getErrorHandlerLoggingLevel() { + return errorHandlerLoggingLevel; + } + + /** + * Allows to configure the default errorHandler logging level for logging uncaught exceptions. + */ + public void setErrorHandlerLoggingLevel(LoggingLevel errorHandlerLoggingLevel) { + this.errorHandlerLoggingLevel = errorHandlerLoggingLevel; + } + + public boolean isErrorHandlerLogStackTrace() { + return errorHandlerLogStackTrace; + } + + /** + * Allows to control whether stacktraces should be logged or not, by the default errorHandler. + */ + public void setErrorHandlerLogStackTrace(boolean errorHandlerLogStackTrace) { + this.errorHandlerLogStackTrace = errorHandlerLogStackTrace; + } } http://git-wip-us.apache.org/repos/asf/camel/blob/7e9047af/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsLoggingExceptionListener.java ---------------------------------------------------------------------- diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsLoggingExceptionListener.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsLoggingExceptionListener.java new file mode 100644 index 0000000..488122d --- /dev/null +++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsLoggingExceptionListener.java @@ -0,0 +1,45 @@ +/** + * 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.component.sjms; + +import javax.jms.ExceptionListener; +import javax.jms.JMSException; + +import org.apache.camel.spi.ExceptionHandler; + +/** + * A JMS {@link ExceptionListener} which logs all caught exceptions. + */ +public class SjmsLoggingExceptionListener implements ExceptionListener { + + private final ExceptionHandler handler; + private final boolean logStackTrace; + + public SjmsLoggingExceptionListener(ExceptionHandler exceptionHandler, boolean logStackTrace) { + this.handler = exceptionHandler; + this.logStackTrace = logStackTrace; + } + + @Override + public void onException(JMSException throwable) { + if (logStackTrace) { + handler.handleException("Execution of JMS message listener failed", throwable); + } else { + handler.handleException("Execution of JMS message listener failed. Caused by: [" + throwable.getMessage() + "]", null); + } + } +} http://git-wip-us.apache.org/repos/asf/camel/blob/7e9047af/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsProducer.java ---------------------------------------------------------------------- diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsProducer.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsProducer.java index 585a6c4..492143d 100644 --- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsProducer.java +++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsProducer.java @@ -287,7 +287,7 @@ public abstract class SjmsProducer extends DefaultAsyncProducer { protected ConnectionResource getOrCreateConnectionResource() { ConnectionResource answer = getEndpoint().getConnectionResource(); if (answer == null) { - answer = getEndpoint().createConnectionResource(); + answer = getEndpoint().createConnectionResource(this); } return answer; } http://git-wip-us.apache.org/repos/asf/camel/blob/7e9047af/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/ConnectionFactoryResource.java ---------------------------------------------------------------------- diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/ConnectionFactoryResource.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/ConnectionFactoryResource.java index d7d7b40..be4159b 100644 --- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/ConnectionFactoryResource.java +++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/ConnectionFactoryResource.java @@ -18,6 +18,7 @@ package org.apache.camel.component.sjms.jms; import javax.jms.Connection; import javax.jms.ConnectionFactory; +import javax.jms.ExceptionListener; import org.apache.camel.util.ObjectHelper; import org.apache.commons.pool.BasePoolableObjectFactory; @@ -35,6 +36,7 @@ public class ConnectionFactoryResource extends BasePoolableObjectFactory<Connect private String username; private String password; private String clientId; + private ExceptionListener exceptionListener; /** * Default Constructor @@ -92,6 +94,10 @@ public class ConnectionFactoryResource extends BasePoolableObjectFactory<Connect if (ObjectHelper.isNotEmpty(getClientId())) { connection.setClientID(getClientId()); } + // we want to listen for exceptions + if (exceptionListener != null) { + connection.setExceptionListener(exceptionListener); + } connection.start(); } return connection; @@ -103,7 +109,6 @@ public class ConnectionFactoryResource extends BasePoolableObjectFactory<Connect connection.stop(); connection.close(); } - } public ConnectionFactory getConnectionFactory() { @@ -138,6 +143,14 @@ public class ConnectionFactoryResource extends BasePoolableObjectFactory<Connect this.clientId = clientId; } + public ExceptionListener getExceptionListener() { + return exceptionListener; + } + + public void setExceptionListener(ExceptionListener exceptionListener) { + this.exceptionListener = exceptionListener; + } + public int size() { return connections.getNumActive() + connections.getNumIdle(); } http://git-wip-us.apache.org/repos/asf/camel/blob/7e9047af/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/manual/ManualFromQueueTest.java ---------------------------------------------------------------------- diff --git a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/manual/ManualFromQueueTest.java b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/manual/ManualFromQueueTest.java new file mode 100644 index 0000000..7e48af7 --- /dev/null +++ b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/manual/ManualFromQueueTest.java @@ -0,0 +1,68 @@ +/** + * 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.component.sjms.manual; + +import java.util.concurrent.TimeUnit; + +import org.apache.activemq.ActiveMQConnectionFactory; +import org.apache.camel.CamelContext; +import org.apache.camel.RoutesBuilder; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.sjms.SjmsComponent; +import org.apache.camel.test.junit4.CamelTestSupport; +import org.junit.Ignore; +import org.junit.Test; + +@Ignore("Manual test") +public class ManualFromQueueTest extends CamelTestSupport { + + // using failover will automatic re-connect with ActiveMQ + // private String url = "failover:tcp://localhost:61616"; + private String url = "tcp://localhost:61616"; + + @Override + protected CamelContext createCamelContext() throws Exception { + CamelContext camel = super.createCamelContext(); + + SjmsComponent sjms = new SjmsComponent(); + log.info("Using live connection to existing ActiveMQ broker running on {}", url); + sjms.setConnectionFactory(new ActiveMQConnectionFactory(url)); + + camel.addComponent("sjms", sjms); + + return camel; + } + + @Test + public void testConsume() throws Exception { + getMockEndpoint("mock:foo").expectedMinimumMessageCount(3); + + assertMockEndpointsSatisfied(1, TimeUnit.MINUTES); + } + + @Override + protected RoutesBuilder createRouteBuilder() throws Exception { + return new RouteBuilder() { + @Override + public void configure() throws Exception { + from("sjms:queue:foo?asyncStartListener=true") + .to("log:foo") + .to("mock:foo"); + } + }; + } +}
