You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2016/12/22 12:26:46 UTC
camel git commit: CAMEL-10643: camel-sjms - Should log caught JMS
exceptions by default.
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 <da...@apache.org>
Authored: Thu Dec 22 13:19:02 2016 +0100
Committer: Claus Ibsen <da...@apache.org>
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");
+ }
+ };
+ }
+}