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 2015/05/06 12:37:42 UTC
[1/5] camel git commit: CAMEL-8711: Expose JMS session to Camel
Message so end users can access it and use it for client ack mode etc.
Repository: camel
Updated Branches:
refs/heads/master e816247d6 -> e319c9e08
CAMEL-8711: Expose JMS session to Camel Message so end users can access it and use it for client ack mode etc.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/e00e0d65
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/e00e0d65
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/e00e0d65
Branch: refs/heads/master
Commit: e00e0d6599b01733c270f3053e23118d35ea0881
Parents: e816247
Author: Claus Ibsen <da...@apache.org>
Authored: Wed May 6 11:20:43 2015 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Wed May 6 11:20:43 2015 +0200
----------------------------------------------------------------------
.../jms/DefaultQueueBrowseStrategy.java | 4 +--
.../component/jms/EndpointMessageListener.java | 12 +++++---
.../apache/camel/component/jms/JmsEndpoint.java | 5 +--
.../apache/camel/component/jms/JmsMessage.java | 32 ++++++++++++++------
.../camel/component/jms/JmsPollingConsumer.java | 3 +-
.../component/jms/reply/QueueReplyManager.java | 4 +--
.../camel/component/jms/reply/ReplyHandler.java | 6 ++--
.../camel/component/jms/reply/ReplyHolder.java | 17 +++++++++--
.../camel/component/jms/reply/ReplyManager.java | 4 +--
.../jms/reply/ReplyManagerSupport.java | 12 +++++---
.../jms/reply/TemporaryQueueReplyHandler.java | 5 +--
.../jms/reply/TemporaryQueueReplyManager.java | 4 +--
12 files changed, 72 insertions(+), 36 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/e00e0d65/components/camel-jms/src/main/java/org/apache/camel/component/jms/DefaultQueueBrowseStrategy.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/DefaultQueueBrowseStrategy.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/DefaultQueueBrowseStrategy.java
index 3258d61..45f5847 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/DefaultQueueBrowseStrategy.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/DefaultQueueBrowseStrategy.java
@@ -51,7 +51,7 @@ public class DefaultQueueBrowseStrategy implements QueueBrowseStrategy {
Enumeration<?> iter = browser.getEnumeration();
for (int i = 0; i < size && iter.hasMoreElements(); i++) {
Message message = (Message) iter.nextElement();
- Exchange exchange = endpoint.createExchange(message);
+ Exchange exchange = endpoint.createExchange(message, session);
answer.add(exchange);
}
return answer;
@@ -71,7 +71,7 @@ public class DefaultQueueBrowseStrategy implements QueueBrowseStrategy {
Enumeration<?> iter = browser.getEnumeration();
for (int i = 0; i < size && iter.hasMoreElements(); i++) {
Message message = (Message) iter.nextElement();
- Exchange exchange = endpoint.createExchange(message);
+ Exchange exchange = endpoint.createExchange(message, session);
answer.add(exchange);
}
return answer;
http://git-wip-us.apache.org/repos/asf/camel/blob/e00e0d65/components/camel-jms/src/main/java/org/apache/camel/component/jms/EndpointMessageListener.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/EndpointMessageListener.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/EndpointMessageListener.java
index f2afc60..3980128 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/EndpointMessageListener.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/EndpointMessageListener.java
@@ -35,6 +35,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jms.core.JmsOperations;
import org.springframework.jms.core.MessageCreator;
+import org.springframework.jms.listener.SessionAwareMessageListener;
import static org.apache.camel.util.ObjectHelper.wrapRuntimeCamelException;
@@ -46,7 +47,7 @@ import static org.apache.camel.util.ObjectHelper.wrapRuntimeCamelException;
*
* @version
*/
-public class EndpointMessageListener implements MessageListener {
+public class EndpointMessageListener implements SessionAwareMessageListener {
private static final Logger LOG = LoggerFactory.getLogger(EndpointMessageListener.class);
private final JmsEndpoint endpoint;
private final AsyncProcessor processor;
@@ -62,7 +63,8 @@ public class EndpointMessageListener implements MessageListener {
this.processor = AsyncProcessorConverterHelper.convert(processor);
}
- public void onMessage(final Message message) {
+ @Override
+ public void onMessage(Message message, Session session) throws JMSException {
LOG.trace("onMessage START");
LOG.debug("{} consumer received JMS message: {}", endpoint, message);
@@ -75,7 +77,7 @@ public class EndpointMessageListener implements MessageListener {
// and disableReplyTo hasn't been explicit enabled
sendReply = replyDestination != null && !disableReplyTo;
- final Exchange exchange = createExchange(message, replyDestination);
+ final Exchange exchange = createExchange(message, session, replyDestination);
if (eagerLoadingOfProperties) {
exchange.getIn().getHeaders();
}
@@ -233,11 +235,11 @@ public class EndpointMessageListener implements MessageListener {
}
}
- public Exchange createExchange(Message message, Object replyDestination) {
+ public Exchange createExchange(Message message, Session session, Object replyDestination) {
Exchange exchange = endpoint.createExchange();
JmsBinding binding = getBinding();
exchange.setProperty(Exchange.BINDING, binding);
- exchange.setIn(new JmsMessage(message, binding));
+ exchange.setIn(new JmsMessage(message, session, binding));
// lets set to an InOut if we have some kind of reply-to destination
if (replyDestination != null && !disableReplyTo) {
http://git-wip-us.apache.org/repos/asf/camel/blob/e00e0d65/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java
index ff9bc32..6389226 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java
@@ -25,6 +25,7 @@ import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Queue;
+import javax.jms.Session;
import javax.jms.TemporaryQueue;
import javax.jms.TemporaryTopic;
import javax.jms.Topic;
@@ -292,9 +293,9 @@ public class JmsEndpoint extends DefaultEndpoint implements HeaderFilterStrategy
return exchange;
}
- public Exchange createExchange(Message message) {
+ public Exchange createExchange(Message message, Session session) {
Exchange exchange = createExchange(getExchangePattern());
- exchange.setIn(new JmsMessage(message, getBinding()));
+ exchange.setIn(new JmsMessage(message, session, getBinding()));
return exchange;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/e00e0d65/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsMessage.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsMessage.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsMessage.java
index 45274cf..1dfbd48 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsMessage.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsMessage.java
@@ -22,6 +22,7 @@ import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Queue;
+import javax.jms.Session;
import javax.jms.Topic;
import org.apache.camel.RuntimeExchangeException;
@@ -39,10 +40,12 @@ import org.slf4j.LoggerFactory;
public class JmsMessage extends DefaultMessage {
private static final Logger LOG = LoggerFactory.getLogger(JmsMessage.class);
private Message jmsMessage;
+ private Session jmsSession;
private JmsBinding binding;
- public JmsMessage(Message jmsMessage, JmsBinding binding) {
+ public JmsMessage(Message jmsMessage, Session jmsSession, JmsBinding binding) {
setJmsMessage(jmsMessage);
+ setJmsSession(jmsSession);
setBinding(binding);
}
@@ -98,13 +101,6 @@ public class JmsMessage extends DefaultMessage {
}
}
- /**
- * Returns the underlying JMS message
- */
- public Message getJmsMessage() {
- return jmsMessage;
- }
-
public JmsBinding getBinding() {
if (binding == null) {
binding = ExchangeHelper.getBinding(getExchange(), JmsBinding.class);
@@ -116,6 +112,13 @@ public class JmsMessage extends DefaultMessage {
this.binding = binding;
}
+ /**
+ * Returns the underlying JMS message
+ */
+ public Message getJmsMessage() {
+ return jmsMessage;
+ }
+
public void setJmsMessage(Message jmsMessage) {
if (jmsMessage != null) {
try {
@@ -127,6 +130,17 @@ public class JmsMessage extends DefaultMessage {
this.jmsMessage = jmsMessage;
}
+ /**
+ * Returns the underlying JMS session
+ */
+ public Session getJmsSession() {
+ return jmsSession;
+ }
+
+ public void setJmsSession(Session jmsSession) {
+ this.jmsSession = jmsSession;
+ }
+
@Override
public void setBody(Object body) {
super.setBody(body);
@@ -186,7 +200,7 @@ public class JmsMessage extends DefaultMessage {
@Override
public JmsMessage newInstance() {
- return new JmsMessage(null, binding);
+ return new JmsMessage(null, null, binding);
}
/**
http://git-wip-us.apache.org/repos/asf/camel/blob/e00e0d65/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsPollingConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsPollingConsumer.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsPollingConsumer.java
index 2df30cf..cbb74b0 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsPollingConsumer.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsPollingConsumer.java
@@ -56,6 +56,7 @@ public class JmsPollingConsumer extends PollingConsumerSupport implements Servic
}
public Exchange receive(long timeout) {
+ // TODO: use api so we can get hold of session
setReceiveTimeout(timeout);
Message message;
// using the selector
@@ -65,7 +66,7 @@ public class JmsPollingConsumer extends PollingConsumerSupport implements Servic
message = template.receive();
}
if (message != null) {
- return getEndpoint().createExchange(message);
+ return getEndpoint().createExchange(message, null);
}
return null;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/e00e0d65/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/QueueReplyManager.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/QueueReplyManager.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/QueueReplyManager.java
index 07ddfad..1b72680 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/QueueReplyManager.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/QueueReplyManager.java
@@ -65,7 +65,7 @@ public class QueueReplyManager extends ReplyManagerSupport {
correlation.put(newCorrelationId, handler, requestTimeout);
}
- protected void handleReplyMessage(String correlationID, Message message) {
+ protected void handleReplyMessage(String correlationID, Message message, Session session) {
ReplyHandler handler = correlation.get(correlationID);
if (handler == null && endpoint.isUseMessageIDAsCorrelationID()) {
handler = waitForProvisionCorrelationToBeUpdated(correlationID, message);
@@ -73,7 +73,7 @@ public class QueueReplyManager extends ReplyManagerSupport {
if (handler != null) {
correlation.remove(correlationID);
- handler.onReply(correlationID, message);
+ handler.onReply(correlationID, message, session);
} else {
// we could not correlate the received reply message to a matching request and therefore
// we cannot continue routing the unknown message
http://git-wip-us.apache.org/repos/asf/camel/blob/e00e0d65/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyHandler.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyHandler.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyHandler.java
index 5579745..e05b5c4 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyHandler.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyHandler.java
@@ -17,6 +17,7 @@
package org.apache.camel.component.jms.reply;
import javax.jms.Message;
+import javax.jms.Session;
/**
* Handles a reply.
@@ -29,9 +30,10 @@ public interface ReplyHandler {
* The reply message was received
*
* @param correlationId the correlation id
- * @param reply the reply message
+ * @param reply the JMS reply message
+ * @param session the JMS session
*/
- void onReply(String correlationId, Message reply);
+ void onReply(String correlationId, Message reply, Session session);
/**
* The reply message was not received and a timeout triggered
http://git-wip-us.apache.org/repos/asf/camel/blob/e00e0d65/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyHolder.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyHolder.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyHolder.java
index a967a06..84945ad 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyHolder.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyHolder.java
@@ -17,6 +17,7 @@
package org.apache.camel.component.jms.reply;
import javax.jms.Message;
+import javax.jms.Session;
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
@@ -32,6 +33,7 @@ public class ReplyHolder {
private final Exchange exchange;
private final AsyncCallback callback;
private final Message message;
+ private final Session session;
private final String originalCorrelationId;
private final String correlationId;
private long timeout;
@@ -40,12 +42,13 @@ public class ReplyHolder {
* Constructor to use when a reply message was received
*/
public ReplyHolder(Exchange exchange, AsyncCallback callback, String originalCorrelationId,
- String correlationId, Message message) {
+ String correlationId, Message message, Session session) {
this.exchange = exchange;
this.callback = callback;
this.originalCorrelationId = originalCorrelationId;
this.correlationId = correlationId;
this.message = message;
+ this.session = session;
}
/**
@@ -53,7 +56,7 @@ public class ReplyHolder {
*/
public ReplyHolder(Exchange exchange, AsyncCallback callback, String originalCorrelationId,
String correlationId, long timeout) {
- this(exchange, callback, originalCorrelationId, correlationId, null);
+ this(exchange, callback, originalCorrelationId, correlationId, null, null);
this.timeout = timeout;
}
@@ -95,6 +98,16 @@ public class ReplyHolder {
}
/**
+ * Gets the JMS session from the received message
+ *
+ * @return the JMS session, or <tt>null</tt> if timeout occurred and no message has been received
+ * @see #isTimeout()
+ */
+ public Session getSession() {
+ return session;
+ }
+
+ /**
* Whether timeout triggered or not.
* <p/>
* A timeout is triggered if <tt>requestTimeout</tt> option has been configured, and a reply message has <b>not</b> been
http://git-wip-us.apache.org/repos/asf/camel/blob/e00e0d65/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManager.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManager.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManager.java
index 9eb0085..e3b65aa 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManager.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManager.java
@@ -20,11 +20,11 @@ import java.util.concurrent.ScheduledExecutorService;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
-import javax.jms.MessageListener;
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
import org.apache.camel.component.jms.JmsEndpoint;
+import org.springframework.jms.listener.SessionAwareMessageListener;
/**
* The {@link ReplyManager} is responsible for handling <a href="http://camel.apache.org/request-reply.html">request-reply</a>
@@ -32,7 +32,7 @@ import org.apache.camel.component.jms.JmsEndpoint;
*
* @version
*/
-public interface ReplyManager extends MessageListener {
+public interface ReplyManager extends SessionAwareMessageListener {
/**
* Sets the belonging {@link org.apache.camel.component.jms.JmsEndpoint}.
http://git-wip-us.apache.org/repos/asf/camel/blob/e00e0d65/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManagerSupport.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManagerSupport.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManagerSupport.java
index 83a7729..a2f70c3 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManagerSupport.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManagerSupport.java
@@ -23,6 +23,7 @@ import java.util.concurrent.TimeUnit;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
+import javax.jms.Session;
import org.apache.camel.AsyncCallback;
import org.apache.camel.CamelContext;
@@ -114,7 +115,7 @@ public abstract class ReplyManagerSupport extends ServiceSupport implements Repl
protected abstract ReplyHandler createReplyHandler(ReplyManager replyManager, Exchange exchange, AsyncCallback callback,
String originalCorrelationId, String correlationId, long requestTimeout);
- public void onMessage(Message message) {
+ public void onMessage(Message message, Session session) throws JMSException {
String correlationID = null;
try {
correlationID = message.getJMSCorrelationID();
@@ -129,14 +130,13 @@ public abstract class ReplyManagerSupport extends ServiceSupport implements Repl
log.debug("Received reply message with correlationID [{}] -> {}", correlationID, message);
// handle the reply message
- handleReplyMessage(correlationID, message);
+ handleReplyMessage(correlationID, message, session);
}
public void processReply(ReplyHolder holder) {
if (holder != null && isRunAllowed()) {
try {
Exchange exchange = holder.getExchange();
- Message message = holder.getMessage();
boolean timeout = holder.isTimeout();
if (timeout) {
@@ -151,7 +151,9 @@ public abstract class ReplyManagerSupport extends ServiceSupport implements Repl
String msg = "reply message with correlationID: " + holder.getCorrelationId() + " not received on destination: " + replyTo;
exchange.setException(new ExchangeTimedOutException(exchange, holder.getRequestTimeout(), msg));
} else {
- JmsMessage response = new JmsMessage(message, endpoint.getBinding());
+ Message message = holder.getMessage();
+ Session session = holder.getSession();
+ JmsMessage response = new JmsMessage(message, session, endpoint.getBinding());
// the JmsBinding is designed to be "pull-based": it will populate the Camel message on demand
// therefore, we link Exchange and OUT message before continuing, so that the JmsBinding has full access
// to everything it may need, and can populate headers, properties, etc. accordingly (solves CAMEL-6218).
@@ -181,7 +183,7 @@ public abstract class ReplyManagerSupport extends ServiceSupport implements Repl
}
}
- protected abstract void handleReplyMessage(String correlationID, Message message);
+ protected abstract void handleReplyMessage(String correlationID, Message message, Session session);
protected abstract AbstractMessageListenerContainer createListenerContainer() throws Exception;
http://git-wip-us.apache.org/repos/asf/camel/blob/e00e0d65/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyHandler.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyHandler.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyHandler.java
index a2b28a2..f752a05 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyHandler.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyHandler.java
@@ -17,6 +17,7 @@
package org.apache.camel.component.jms.reply;
import javax.jms.Message;
+import javax.jms.Session;
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
@@ -47,9 +48,9 @@ public class TemporaryQueueReplyHandler implements ReplyHandler {
this.timeout = timeout;
}
- public void onReply(String correlationId, Message reply) {
+ public void onReply(String correlationId, Message reply, Session session) {
// create holder object with the the reply
- ReplyHolder holder = new ReplyHolder(exchange, callback, originalCorrelationId, correlationId, reply);
+ ReplyHolder holder = new ReplyHolder(exchange, callback, originalCorrelationId, correlationId, reply, session);
// process the reply
replyManager.processReply(holder);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/e00e0d65/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyManager.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyManager.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyManager.java
index 0e3d98b..6d1f51d 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyManager.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyManager.java
@@ -72,7 +72,7 @@ public class TemporaryQueueReplyManager extends ReplyManagerSupport {
}
@Override
- protected void handleReplyMessage(String correlationID, Message message) {
+ protected void handleReplyMessage(String correlationID, Message message, Session session) {
ReplyHandler handler = correlation.get(correlationID);
if (handler == null && endpoint.isUseMessageIDAsCorrelationID()) {
handler = waitForProvisionCorrelationToBeUpdated(correlationID, message);
@@ -80,7 +80,7 @@ public class TemporaryQueueReplyManager extends ReplyManagerSupport {
if (handler != null) {
correlation.remove(correlationID);
- handler.onReply(correlationID, message);
+ handler.onReply(correlationID, message, session);
} else {
// we could not correlate the received reply message to a matching request and therefore
// we cannot continue routing the unknown message
[4/5] camel git commit: CAMEL-8711: Lets be compatible with older
activemq-camel versions.
Posted by da...@apache.org.
CAMEL-8711: Lets be compatible with older activemq-camel versions.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/7302e8f2
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/7302e8f2
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/7302e8f2
Branch: refs/heads/master
Commit: 7302e8f2648b82ffa2ea23a11d3bb2ecf3711411
Parents: a8949a8
Author: Claus Ibsen <da...@apache.org>
Authored: Wed May 6 12:35:52 2015 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Wed May 6 12:35:52 2015 +0200
----------------------------------------------------------------------
.../main/java/org/apache/camel/component/jms/JmsMessage.java | 5 +++++
1 file changed, 5 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/7302e8f2/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsMessage.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsMessage.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsMessage.java
index 1dfbd48..964cebe 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsMessage.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsMessage.java
@@ -43,6 +43,11 @@ public class JmsMessage extends DefaultMessage {
private Session jmsSession;
private JmsBinding binding;
+ @Deprecated
+ public JmsMessage(Message jmsMessage, JmsBinding binding) {
+ this(jmsMessage, null, binding);
+ }
+
public JmsMessage(Message jmsMessage, Session jmsSession, JmsBinding binding) {
setJmsMessage(jmsMessage);
setJmsSession(jmsSession);
[5/5] camel git commit: CAMEL-8711: Expose JMS session to Camel
Message so end users can access it and use it for client ack mode etc.
Posted by da...@apache.org.
CAMEL-8711: Expose JMS session to Camel Message so end users can access it and use it for client ack mode etc.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/e319c9e0
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/e319c9e0
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/e319c9e0
Branch: refs/heads/master
Commit: e319c9e08cce71e7757f112f9fcb9ce9583da817
Parents: 7302e8f
Author: Claus Ibsen <da...@apache.org>
Authored: Wed May 6 12:41:12 2015 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Wed May 6 12:41:12 2015 +0200
----------------------------------------------------------------------
.../main/java/org/apache/camel/component/jms/JmsMessage.java | 5 ++++-
1 file changed, 4 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/e319c9e0/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsMessage.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsMessage.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsMessage.java
index 964cebe..c75cdde 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsMessage.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsMessage.java
@@ -136,7 +136,10 @@ public class JmsMessage extends DefaultMessage {
}
/**
- * Returns the underlying JMS session
+ * Returns the underlying JMS session.
+ * <p/>
+ * This may be <tt>null</tt> if using {@link org.apache.camel.component.jms.JmsPollingConsumer},
+ * or the broker component from Apache ActiveMQ 5.11.x or older.
*/
public Session getJmsSession() {
return jmsSession;
[3/5] camel git commit: Polished
Posted by da...@apache.org.
Polished
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/a8949a83
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/a8949a83
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/a8949a83
Branch: refs/heads/master
Commit: a8949a83f5d4abb53105b016bf7a88ad63401d6a
Parents: e82caec
Author: Claus Ibsen <da...@apache.org>
Authored: Wed May 6 11:53:28 2015 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Wed May 6 12:01:58 2015 +0200
----------------------------------------------------------------------
.../java/org/apache/camel/component/jms/JmsConfiguration.java | 4 ----
.../java/org/apache/camel/component/jms/JmsPollingConsumer.java | 1 -
.../org/apache/camel/component/jms/issues/JmsHammerTest.java | 3 +--
3 files changed, 1 insertion(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/a8949a83/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java
index 58a290c..e372293 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java
@@ -1133,10 +1133,6 @@ public class JmsConfiguration implements Cloneable {
listener.setReplyToDestination(getReplyTo());
}
- // TODO: REVISIT: We really ought to change the model and let JmsProducer
- // and JmsConsumer have their own JmsConfiguration instance
- // This way producer's and consumer's QoS can differ and be
- // independently configured
JmsOperations operations = listener.getTemplate();
if (operations instanceof JmsTemplate) {
JmsTemplate template = (JmsTemplate) operations;
http://git-wip-us.apache.org/repos/asf/camel/blob/a8949a83/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsPollingConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsPollingConsumer.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsPollingConsumer.java
index cbb74b0..1ad9cdc 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsPollingConsumer.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsPollingConsumer.java
@@ -56,7 +56,6 @@ public class JmsPollingConsumer extends PollingConsumerSupport implements Servic
}
public Exchange receive(long timeout) {
- // TODO: use api so we can get hold of session
setReceiveTimeout(timeout);
Message message;
// using the selector
http://git-wip-us.apache.org/repos/asf/camel/blob/a8949a83/components/camel-jms/src/test/java/org/apache/camel/component/jms/issues/JmsHammerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/test/java/org/apache/camel/component/jms/issues/JmsHammerTest.java b/components/camel-jms/src/test/java/org/apache/camel/component/jms/issues/JmsHammerTest.java
index 36a0d72..07076b3 100644
--- a/components/camel-jms/src/test/java/org/apache/camel/component/jms/issues/JmsHammerTest.java
+++ b/components/camel-jms/src/test/java/org/apache/camel/component/jms/issues/JmsHammerTest.java
@@ -53,8 +53,7 @@ public class JmsHammerTest extends CamelTestSupport {
protected CamelContext createCamelContext() throws Exception {
CamelContext camelContext = super.createCamelContext();
- // TODO: MR-170
- ActiveMQComponent activemq =
+ ActiveMQComponent activemq =
activeMQComponent("vm://localhost?broker.persistent=false&broker.useJmx=false&jms.redeliveryPolicy.maximumRedeliveries=0"
+ "&jms.redeliveryPolicy.initialRedeliveryDelay=500&jms.useAsyncSend=false&jms.sendTimeout=10000"
+ "&jms.maxReconnectAttempts=1&jms.timeout=3000");
[2/5] camel git commit: CAMEL-8711: Expose JMS session to Camel
Message so end users can access it and use it for client ack mode etc.
Posted by da...@apache.org.
CAMEL-8711: Expose JMS session to Camel Message so end users can access it and use it for client ack mode etc.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/e82caec2
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/e82caec2
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/e82caec2
Branch: refs/heads/master
Commit: e82caec2216f0b6f7aaf382aa7be70a37a902aa5
Parents: e00e0d6
Author: Claus Ibsen <da...@apache.org>
Authored: Wed May 6 11:26:49 2015 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Wed May 6 11:26:49 2015 +0200
----------------------------------------------------------------------
.../camel/component/jms/JmsClientAckTest.java | 80 ++++++++++++++++++++
.../camel/component/jms/JmsSessionTest.java | 77 +++++++++++++++++++
2 files changed, 157 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/e82caec2/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsClientAckTest.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsClientAckTest.java b/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsClientAckTest.java
new file mode 100644
index 0000000..3ba22dd
--- /dev/null
+++ b/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsClientAckTest.java
@@ -0,0 +1,80 @@
+/**
+ * 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.jms;
+
+import javax.jms.ConnectionFactory;
+import javax.jms.Session;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+
+import static org.apache.camel.component.jms.JmsComponent.jmsComponentClientAcknowledge;
+
+/**
+ * @version
+ */
+public class JmsClientAckTest extends CamelTestSupport {
+
+ protected String componentName = "activemq";
+
+ @Test
+ public void testJmsClientAck() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
+
+ template.sendBody("activemq:queue:foo", "Hello World");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+
+ ConnectionFactory connectionFactory = CamelJmsTestHelper.createConnectionFactory();
+ camelContext.addComponent(componentName, jmsComponentClientAcknowledge(connectionFactory));
+
+ return camelContext;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("activemq:queue:foo")
+ .process(new Processor() {
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ JmsMessage jms = exchange.getIn(JmsMessage.class);
+ assertNotNull(jms);
+ Session session = jms.getJmsSession();
+ assertNotNull("Should have JMS session", session);
+
+ assertEquals("Should be client ACK mode", Session.CLIENT_ACKNOWLEDGE, session.getAcknowledgeMode());
+ jms.getJmsMessage().acknowledge();
+ }
+ })
+ .to("mock:result");
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e82caec2/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsSessionTest.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsSessionTest.java b/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsSessionTest.java
new file mode 100644
index 0000000..48ddaa0
--- /dev/null
+++ b/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsSessionTest.java
@@ -0,0 +1,77 @@
+/**
+ * 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.jms;
+
+import javax.jms.ConnectionFactory;
+import javax.jms.Session;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+
+import static org.apache.camel.component.jms.JmsComponent.jmsComponentAutoAcknowledge;
+
+/**
+ * @version
+ */
+public class JmsSessionTest extends CamelTestSupport {
+
+ protected String componentName = "activemq";
+
+ @Test
+ public void testJmsSession() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
+
+ template.sendBody("activemq:queue:foo", "Hello World");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+
+ ConnectionFactory connectionFactory = CamelJmsTestHelper.createConnectionFactory();
+ camelContext.addComponent(componentName, jmsComponentAutoAcknowledge(connectionFactory));
+
+ return camelContext;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("activemq:queue:foo")
+ .process(new Processor() {
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ JmsMessage jms = exchange.getIn(JmsMessage.class);
+ assertNotNull(jms);
+ Session session = jms.getJmsSession();
+ assertNotNull("Should have JMS session", session);
+ }
+ })
+ .to("mock:result");
+ }
+ };
+ }
+}