You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by su...@apache.org on 2013/10/30 14:21:58 UTC
[1/3] git commit: CAMEL-6362: Consumers should always use dedicated
Sessions
Updated Branches:
refs/heads/master 0ab6bba5b -> dbe794c37
CAMEL-6362: Consumers should always use dedicated Sessions
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/520e55fa
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/520e55fa
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/520e55fa
Branch: refs/heads/master
Commit: 520e55fa81f7aa9faa65960ad36be0af8f3d03d0
Parents: 29e4411
Author: Scott England-Sullivan <su...@apache.org>
Authored: Tue Oct 29 11:17:35 2013 -0500
Committer: Scott England-Sullivan <su...@apache.org>
Committed: Tue Oct 29 11:17:35 2013 -0500
----------------------------------------------------------------------
.../camel/component/sjms/SjmsConsumer.java | 75 +++++------
.../camel/component/sjms/SjmsEndpoint.java | 42 +-----
.../camel/component/sjms/SjmsProducer.java | 5 +-
.../component/sjms/producer/InOnlyProducer.java | 102 +++++++++------
.../component/sjms/producer/InOutProducer.java | 127 ++++++++++++-------
.../InOnlyTopicDurableConsumerTest.java | 4 +-
6 files changed, 180 insertions(+), 175 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/520e55fa/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 806537d..5b01c3a 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
@@ -23,6 +23,7 @@ import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
+import org.apache.camel.CamelException;
import org.apache.camel.Endpoint;
import org.apache.camel.ExchangePattern;
import org.apache.camel.Processor;
@@ -32,7 +33,6 @@ import org.apache.camel.component.sjms.consumer.InOutMessageHandler;
import org.apache.camel.component.sjms.jms.ConnectionResource;
import org.apache.camel.component.sjms.jms.JmsObjectFactory;
import org.apache.camel.component.sjms.jms.ObjectPool;
-import org.apache.camel.component.sjms.jms.SessionPool;
import org.apache.camel.component.sjms.taskmanager.TimedTaskManager;
import org.apache.camel.component.sjms.tx.BatchTransactionCommitStrategy;
import org.apache.camel.component.sjms.tx.DefaultTransactionCommitStrategy;
@@ -66,12 +66,7 @@ public class SjmsConsumer extends DefaultConsumer {
*/
@Override
protected MessageConsumerResources createObject() throws Exception {
- MessageConsumerResources model = null;
- if (isTransacted() || getEndpoint().getExchangePattern().equals(ExchangePattern.InOut)) {
- model = createConsumerWithDedicatedSession();
- } else {
- model = createConsumerListener();
- }
+ MessageConsumerResources model = createConsumer();
return model;
}
@@ -168,39 +163,41 @@ public class SjmsConsumer extends DefaultConsumer {
* Creates a {@link MessageConsumerResources} with a dedicated
* {@link Session} required for transacted and InOut consumers.
*/
- private MessageConsumerResources createConsumerWithDedicatedSession() throws Exception {
- Connection conn = getConnectionResource().borrowConnection();
- Session session = null;
- if (isTransacted()) {
- session = conn.createSession(true, Session.SESSION_TRANSACTED);
- } else {
- session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ @SuppressWarnings("unused")
+ private MessageConsumerResources createConsumer() throws Exception {
+ MessageConsumerResources answer = null;
+ Connection conn = null;
+ try {
+ conn = getConnectionResource().borrowConnection();
+
+ Session session = null;
+ MessageConsumer messageConsumer = null;
+ if (isTransacted()) {
+ session = conn.createSession(true, Session.SESSION_TRANSACTED);
+ } else {
+ session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ }
+ messageConsumer = JmsObjectFactory.createMessageConsumer(session, getDestinationName(), getMessageSelector(), isTopic(), getDurableSubscriptionId());
+ MessageListener handler = createMessageHandler(session);
+ messageConsumer.setMessageListener(handler);
+
+ if (session == null) {
+ throw new CamelException("Message Consumer Creation Exception: Session is NULL");
+ }
+ if (messageConsumer == null) {
+ throw new CamelException("Message Consumer Creation Exception: MessageConsumer is NULL");
+ }
+ answer = new MessageConsumerResources(session, messageConsumer);
+ } catch (Exception e) {
+ log.error("Unable to create the MessageConsumer: " + e.getLocalizedMessage());
+ } finally {
+ if (conn != null) {
+ getConnectionResource().returnConnection(conn);
+ }
}
- MessageConsumer messageConsumer = JmsObjectFactory.createMessageConsumer(session, getDestinationName(), getMessageSelector(), isTopic(), getDurableSubscriptionId());
- MessageListener handler = createMessageHandler(session);
- messageConsumer.setMessageListener(handler);
- getConnectionResource().returnConnection(conn);
- return new MessageConsumerResources(session, messageConsumer);
+ return answer;
}
- /**
- * Creates a {@link MessageConsumerResources} with a shared {@link Session}
- * for non-transacted InOnly consumers.
- */
- private MessageConsumerResources createConsumerListener() throws Exception {
- Session queueSession = getSessionPool().borrowObject();
- MessageConsumer messageConsumer = null;
- if (isTopic()) {
- messageConsumer = JmsObjectFactory.createTopicConsumer(queueSession, getDestinationName(), getMessageSelector());
- } else {
- messageConsumer = JmsObjectFactory.createQueueConsumer(queueSession, getDestinationName(), getMessageSelector());
- }
- getSessionPool().returnObject(queueSession);
- // Don't pass in the session. Only needed if we are transacted
- MessageListener handler = createMessageHandler(null);
- messageConsumer.setMessageListener(handler);
- return new MessageConsumerResources(messageConsumer);
- }
/**
* Helper factory method used to create a MessageListener based on the MEP
@@ -253,10 +250,6 @@ public class SjmsConsumer extends DefaultConsumer {
return getEndpoint().getConnectionResource();
}
- protected SessionPool getSessionPool() {
- return getEndpoint().getSessions();
- }
-
public int getAcknowledgementMode() {
return getEndpoint().getAcknowledgementMode().intValue();
}
http://git-wip-us.apache.org/repos/asf/camel/blob/520e55fa/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 edaa70c..63118ec 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
@@ -25,7 +25,6 @@ import org.apache.camel.Producer;
import org.apache.camel.component.sjms.jms.ConnectionResource;
import org.apache.camel.component.sjms.jms.KeyFormatStrategy;
import org.apache.camel.component.sjms.jms.SessionAcknowledgementType;
-import org.apache.camel.component.sjms.jms.SessionPool;
import org.apache.camel.component.sjms.producer.InOnlyProducer;
import org.apache.camel.component.sjms.producer.InOutProducer;
import org.apache.camel.impl.DefaultEndpoint;
@@ -42,7 +41,7 @@ import org.slf4j.LoggerFactory;
public class SjmsEndpoint extends DefaultEndpoint implements MultipleConsumersSupport {
protected final Logger logger = LoggerFactory.getLogger(getClass());
- private SessionPool sessions;
+
@UriParam
private boolean synchronous = true;
@UriParam
@@ -95,28 +94,10 @@ public class SjmsEndpoint extends DefaultEndpoint implements MultipleConsumersSu
@Override
protected void doStart() throws Exception {
super.doStart();
-
- //
- // TODO since we only need a session pool for one use case, find a
- // better way
- //
- // We only create a session pool when we are not transacted.
- // Transacted listeners or producers need to be paired with the
- // Session that created them.
- if (!isTransacted() && getExchangePattern().equals(ExchangePattern.InOnly)) {
- sessions = new SessionPool(getSessionCount(), getConnectionResource());
-
- // TODO fix the string hack
- sessions.setAcknowledgeMode(SessionAcknowledgementType.valueOf(getAcknowledgementMode() + ""));
- getSessions().fillPool();
- }
}
@Override
protected void doStop() throws Exception {
- if (getSessions() != null) {
- getSessions().drainPool();
- }
super.doStop();
}
@@ -169,25 +150,6 @@ public class SjmsEndpoint extends DefaultEndpoint implements MultipleConsumersSu
}
/**
- * Returns a SessionPool if available.
- *
- * @return the sessions
- */
- public SessionPool getSessions() {
- return sessions;
- }
-
- /**
- * SessionPool used by endpoints that do not require a dedicated session per
- * consumer or producer.
- *
- * @param sessions default null
- */
- public void setSessions(SessionPool sessions) {
- this.sessions = sessions;
- }
-
- /**
* Use to determine whether or not to process exchanges synchronously.
*
* @return true if endoint is synchronous, otherwise false
@@ -239,6 +201,7 @@ public class SjmsEndpoint extends DefaultEndpoint implements MultipleConsumersSu
*
* @return the sessionCount
*/
+ @Deprecated
public int getSessionCount() {
return sessionCount;
}
@@ -250,6 +213,7 @@ public class SjmsEndpoint extends DefaultEndpoint implements MultipleConsumersSu
*
* @param sessionCount the number of Session instances, default is 1
*/
+ @Deprecated
public void setSessionCount(int sessionCount) {
this.sessionCount = sessionCount;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/520e55fa/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 d4e1a1d..c5d9c1f 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
@@ -157,7 +157,7 @@ public abstract class SjmsProducer extends DefaultAsyncProducer {
if (log.isDebugEnabled()) {
log.debug("Processing Exchange.id:{}", exchange.getExchangeId());
}
-
+
try {
if (!isSynchronous()) {
if (log.isDebugEnabled()) {
@@ -189,10 +189,9 @@ public abstract class SjmsProducer extends DefaultAsyncProducer {
exchange.setException(e);
}
log.debug("Processing Exchange.id:{}", exchange.getExchangeId() + " - SUCCESS");
-
+
return isSynchronous();
}
-
protected SjmsEndpoint getSjmsEndpoint() {
return (SjmsEndpoint)this.getEndpoint();
http://git-wip-us.apache.org/repos/asf/camel/blob/520e55fa/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/producer/InOnlyProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/producer/InOnlyProducer.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/producer/InOnlyProducer.java
index ca49765..df689b2 100644
--- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/producer/InOnlyProducer.java
+++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/producer/InOnlyProducer.java
@@ -51,32 +51,44 @@ public class InOnlyProducer extends SjmsProducer {
*/
@Override
public MessageProducerResources doCreateProducerModel() throws Exception {
- Connection conn = getConnectionResource().borrowConnection();
- TransactionCommitStrategy commitStrategy = null;
- Session session = null;
- if (isEndpointTransacted()) {
- if (getCommitStrategy() != null) {
- commitStrategy = getCommitStrategy();
+ MessageProducerResources answer = null;
+ Connection conn = null;
+ try {
+ conn = getConnectionResource().borrowConnection();
+
+ TransactionCommitStrategy commitStrategy = null;
+ Session session = null;
+ MessageProducer messageProducer = null;
+
+ if (isEndpointTransacted()) {
+ if (getCommitStrategy() != null) {
+ commitStrategy = getCommitStrategy();
+ } else {
+ commitStrategy = new DefaultTransactionCommitStrategy();
+ }
+ session = conn.createSession(true, getAcknowledgeMode());
} else {
- commitStrategy = new DefaultTransactionCommitStrategy();
+ session = conn.createSession(false, getAcknowledgeMode());
+ }
+ if (isTopic()) {
+ messageProducer = JmsObjectFactory.createMessageProducer(session, getDestinationName(), isTopic(), isPersistent(), getTtl());
+ } else {
+ messageProducer = JmsObjectFactory.createQueueProducer(session, getDestinationName());
+ }
+ answer = new MessageProducerResources(session, messageProducer, commitStrategy);
+ } catch (Exception e) {
+ log.error("Unable to create the MessageProducer: " + e.getLocalizedMessage());
+ } finally {
+ if (conn != null) {
+ getConnectionResource().returnConnection(conn);
}
- session = conn.createSession(true, getAcknowledgeMode());
- } else {
- session = conn.createSession(false, getAcknowledgeMode());
- }
- MessageProducer messageProducer = null;
- if (isTopic()) {
- messageProducer = JmsObjectFactory.createMessageProducer(session, getDestinationName(), isTopic(), isPersistent(), getTtl());
- } else {
- messageProducer = JmsObjectFactory.createQueueProducer(session, getDestinationName());
}
- getConnectionResource().returnConnection(conn);
- return new MessageProducerResources(session, messageProducer, commitStrategy);
+ return answer;
}
/*
- * @see org.apache.camel.component.sjms.SjmsProducer#sendMessage(org.apache.camel.Exchange, org.apache.camel.AsyncCallback)
- *
+ * @see
+ * org.apache.camel.component.sjms.SjmsProducer#sendMessage(org.apache.camel.Exchange, org.apache.camel.AsyncCallback)
* @param exchange
* @param callback
* @throws Exception
@@ -85,34 +97,40 @@ public class InOnlyProducer extends SjmsProducer {
public void sendMessage(Exchange exchange, AsyncCallback callback) throws Exception {
List<Message> messages = new ArrayList<Message>();
MessageProducerResources producer = getProducers().borrowObject();
- if (getProducers() != null) {
- if (exchange.getIn().getBody() != null) {
- if (exchange.getIn().getBody() instanceof List) {
- List<?> payload = (List<?>)exchange.getIn().getBody();
- for (Object object : payload) {
- Message message = null;
- if (BatchMessage.class.isInstance(object)) {
- BatchMessage<?> batchMessage = (BatchMessage<?>)object;
- message = JmsMessageHelper.createMessage(producer.getSession(), batchMessage.getPayload(), batchMessage.getHeaders(), getSjmsEndpoint()
- .getJmsKeyFormatStrategy());
- } else {
- message = JmsMessageHelper.createMessage(producer.getSession(), object, exchange.getIn().getHeaders(), getSjmsEndpoint().getJmsKeyFormatStrategy());
+ try {
+ if (getProducers() != null) {
+ if (exchange.getIn().getBody() != null) {
+ if (exchange.getIn().getBody() instanceof List) {
+ List<?> payload = (List<?>)exchange.getIn().getBody();
+ for (Object object : payload) {
+ Message message = null;
+ if (BatchMessage.class.isInstance(object)) {
+ BatchMessage<?> batchMessage = (BatchMessage<?>)object;
+ message = JmsMessageHelper.createMessage(producer.getSession(), batchMessage.getPayload(), batchMessage.getHeaders(), getSjmsEndpoint()
+ .getJmsKeyFormatStrategy());
+ } else {
+ message = JmsMessageHelper.createMessage(producer.getSession(), object, exchange.getIn().getHeaders(), getSjmsEndpoint().getJmsKeyFormatStrategy());
+ }
+ messages.add(message);
}
+ } else {
+ Object payload = exchange.getIn().getBody();
+ Message message = JmsMessageHelper
+ .createMessage(producer.getSession(), payload, exchange.getIn().getHeaders(), getSjmsEndpoint().getJmsKeyFormatStrategy());
messages.add(message);
}
- } else {
- Object payload = exchange.getIn().getBody();
- Message message = JmsMessageHelper.createMessage(producer.getSession(), payload, exchange.getIn().getHeaders(), getSjmsEndpoint().getJmsKeyFormatStrategy());
- messages.add(message);
}
- }
- if (isEndpointTransacted()) {
- exchange.getUnitOfWork().addSynchronization(new SessionTransactionSynchronization(producer.getSession(), producer.getCommitStrategy()));
- }
- for (Message message : messages) {
- producer.getMessageProducer().send(message);
+ if (isEndpointTransacted()) {
+ exchange.getUnitOfWork().addSynchronization(new SessionTransactionSynchronization(producer.getSession(), producer.getCommitStrategy()));
+ }
+ for (Message message : messages) {
+ producer.getMessageProducer().send(message);
+ }
}
+ } catch (Exception e) {
+ exchange.setException(new Exception("Unable to complet sending the message: " + e.getLocalizedMessage()));
+ } finally {
getProducers().returnObject(producer);
callback.done(isSynchronous());
}
http://git-wip-us.apache.org/repos/asf/camel/blob/520e55fa/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/producer/InOutProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/producer/InOutProducer.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/producer/InOutProducer.java
index c98e136..0936ecf 100644
--- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/producer/InOutProducer.java
+++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/producer/InOutProducer.java
@@ -79,40 +79,49 @@ public class InOutProducer extends SjmsProducer {
@Override
protected MessageConsumerResource createObject() throws Exception {
- Connection conn = getConnectionResource().borrowConnection();
+ MessageConsumerResource answer = null;
+ Connection conn = null;
Session session = null;
- if (isEndpointTransacted()) {
- session = conn.createSession(true, Session.SESSION_TRANSACTED);
- } else {
- session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
- }
- getConnectionResource().returnConnection(conn);
- Destination replyToDestination = null;
- if (ObjectHelper.isEmpty(getNamedReplyTo())) {
- replyToDestination = JmsObjectFactory.createTemporaryDestination(session, isTopic());
- } else {
- replyToDestination = JmsObjectFactory.createDestination(session, getNamedReplyTo(), isTopic());
- }
- MessageConsumer messageConsumer = JmsObjectFactory.createMessageConsumer(session, replyToDestination, null, isTopic(), null, true);
- messageConsumer.setMessageListener(new MessageListener() {
-
- @Override
- public void onMessage(Message message) {
- if (logger.isDebugEnabled()) {
- logger.debug("Message Received in the Consumer Pool");
- logger.debug(" Message : {}", message);
- }
- try {
- Exchanger<Object> exchanger = exchangerMap.get(message.getJMSCorrelationID());
- exchanger.exchange(message, getResponseTimeOut(), TimeUnit.MILLISECONDS);
- } catch (Exception e) {
- ObjectHelper.wrapRuntimeCamelException(e);
- }
+ try {
+ conn = getConnectionResource().borrowConnection();
+ if (isEndpointTransacted()) {
+ session = conn.createSession(true, Session.SESSION_TRANSACTED);
+ } else {
+ session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ }
+ Destination replyToDestination = null;
+ if (ObjectHelper.isEmpty(getNamedReplyTo())) {
+ replyToDestination = JmsObjectFactory.createTemporaryDestination(session, isTopic());
+ } else {
+ replyToDestination = JmsObjectFactory.createDestination(session, getNamedReplyTo(), isTopic());
}
- });
- MessageConsumerResource mcm = new MessageConsumerResource(session, messageConsumer, replyToDestination);
- return mcm;
+ MessageConsumer messageConsumer = JmsObjectFactory.createMessageConsumer(session, replyToDestination, null, isTopic(), null, true);
+ messageConsumer.setMessageListener(new MessageListener() {
+
+ @Override
+ public void onMessage(Message message) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Message Received in the Consumer Pool");
+ logger.debug(" Message : {}", message);
+ }
+ try {
+ Exchanger<Object> exchanger = exchangerMap.get(message.getJMSCorrelationID());
+ exchanger.exchange(message, getResponseTimeOut(), TimeUnit.MILLISECONDS);
+ } catch (Exception e) {
+ ObjectHelper.wrapRuntimeCamelException(e);
+ }
+
+ }
+ });
+ answer = new MessageConsumerResource(session, messageConsumer, replyToDestination);
+ } catch (Exception e) {
+ log.error("Unable to create the MessageConsumerResource: " + e.getLocalizedMessage());
+ throw new CamelException(e);
+ } finally {
+ getConnectionResource().returnConnection(conn);
+ }
+ return answer;
}
@Override
@@ -227,21 +236,42 @@ public class InOutProducer extends SjmsProducer {
@Override
public MessageProducerResources doCreateProducerModel() throws Exception {
- Connection conn = getConnectionResource().borrowConnection();
- Session session = null;
- if (isEndpointTransacted()) {
- session = conn.createSession(true, getAcknowledgeMode());
- } else {
- session = conn.createSession(false, getAcknowledgeMode());
- }
- MessageProducer messageProducer = null;
- if (isTopic()) {
- messageProducer = JmsObjectFactory.createMessageProducer(session, getDestinationName(), isTopic(), isPersistent(), getTtl());
- } else {
- messageProducer = JmsObjectFactory.createQueueProducer(session, getDestinationName());
+ MessageProducerResources answer = null;
+ Connection conn = null;
+ try {
+ MessageProducer messageProducer = null;
+ Session session = null;
+
+ conn = getConnectionResource().borrowConnection();
+ if (isEndpointTransacted()) {
+ session = conn.createSession(true, getAcknowledgeMode());
+ } else {
+ session = conn.createSession(false, getAcknowledgeMode());
+ }
+ if (isTopic()) {
+ messageProducer = JmsObjectFactory.createMessageProducer(session, getDestinationName(), isTopic(), isPersistent(), getTtl());
+ } else {
+ messageProducer = JmsObjectFactory.createQueueProducer(session, getDestinationName());
+ }
+
+ if (session == null) {
+ throw new CamelException("Message Consumer Creation Exception: Session is NULL");
+ }
+ if (messageProducer == null) {
+ throw new CamelException("Message Consumer Creation Exception: MessageProducer is NULL");
+ }
+
+ answer = new MessageProducerResources(session, messageProducer);
+
+ } catch (Exception e) {
+ log.error("Unable to create the MessageProducer: " + e.getLocalizedMessage());
+ } finally {
+ if (conn != null) {
+ getConnectionResource().returnConnection(conn);
+ }
}
- getConnectionResource().returnConnection(conn);
- return new MessageProducerResources(session, messageProducer);
+
+ return answer;
}
/**
@@ -269,9 +299,9 @@ public class InOutProducer extends SjmsProducer {
if (isEndpointTransacted()) {
exchange.getUnitOfWork().addSynchronization(new SessionTransactionSynchronization(producer.getSession(), getCommitStrategy()));
}
-
+
Message request = SjmsExchangeMessageHelper.createMessage(exchange, producer.getSession(), getSjmsEndpoint().getJmsKeyFormatStrategy());
-
+
// TODO just set the correlation id don't get it from the
// message
String correlationId = null;
@@ -295,7 +325,8 @@ public class InOutProducer extends SjmsProducer {
consumers.returnObject(consumer);
producer.getMessageProducer().send(request);
- // Return the producer to the pool so another waiting producer can move forward
+ // Return the producer to the pool so another waiting producer
+ // can move forward
// without waiting on us to complete the exchange
try {
getProducers().returnObject(producer);
http://git-wip-us.apache.org/repos/asf/camel/blob/520e55fa/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/InOnlyTopicDurableConsumerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/InOnlyTopicDurableConsumerTest.java b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/InOnlyTopicDurableConsumerTest.java
index 9fa0a80..2420476 100644
--- a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/InOnlyTopicDurableConsumerTest.java
+++ b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/InOnlyTopicDurableConsumerTest.java
@@ -78,10 +78,10 @@ public class InOnlyTopicDurableConsumerTest extends CamelTestSupport {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("sjms:topic:foo?durableSubscriptionId=bar")
+ from("sjms:topic:foo?durableSubscriptionId=bar1")
.to("mock:result");
- from("sjms:topic:foo?durableSubscriptionId=bar")
+ from("sjms:topic:foo?durableSubscriptionId=bar2")
.to("mock:result2");
}
};
[2/3] git commit: Merge branch 'master' of
https://git-wip-us.apache.org/repos/asf/camel
Posted by su...@apache.org.
Merge branch 'master' of https://git-wip-us.apache.org/repos/asf/camel
Merging in latest from upstream
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/9b969680
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/9b969680
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/9b969680
Branch: refs/heads/master
Commit: 9b9696808268cec710e001c1f7fd5fce34b3b3a2
Parents: 520e55f cdad658
Author: Scott England-Sullivan <su...@apache.org>
Authored: Wed Oct 30 08:15:56 2013 -0500
Committer: Scott England-Sullivan <su...@apache.org>
Committed: Wed Oct 30 08:15:56 2013 -0500
----------------------------------------------------------------------
.../camel/component/aws/s3/S3Configuration.java | 10 ++++++
.../camel/component/aws/s3/S3Consumer.java | 38 ++++++++++++++------
.../camel/component/aws/s3/S3Endpoint.java | 9 ++++-
.../component/aws/s3/AmazonS3ClientMock.java | 5 +--
.../camel/component/http4/HttpEndpoint.java | 9 +++++
.../camel/component/http4/HttpProducer.java | 7 ++++
.../component/http4/HttpAuthenticationTest.java | 12 +++++++
7 files changed, 77 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
[3/3] git commit: Merge branch 'master' of
https://git-wip-us.apache.org/repos/asf/camel
Posted by su...@apache.org.
Merge branch 'master' of https://git-wip-us.apache.org/repos/asf/camel
Merge latests from upstream
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/dbe794c3
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/dbe794c3
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/dbe794c3
Branch: refs/heads/master
Commit: dbe794c37ed84053eee81e9a97b0b2eef4fc2e89
Parents: 9b96968 0ab6bba
Author: Scott England-Sullivan <su...@apache.org>
Authored: Wed Oct 30 08:21:13 2013 -0500
Committer: Scott England-Sullivan <su...@apache.org>
Committed: Wed Oct 30 08:21:13 2013 -0500
----------------------------------------------------------------------
.../netty/http/DefaultNettyHttpBinding.java | 30 +++++++++++++++++---
.../netty/http/NettyHttpContentTypeTest.java | 16 +++++++++++
2 files changed, 42 insertions(+), 4 deletions(-)
----------------------------------------------------------------------