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 2012/10/06 00:36:22 UTC
svn commit: r1394845 - in
/camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms:
./ consumer/ jms/ producer/
Author: sully6768
Date: Fri Oct 5 22:36:20 2012
New Revision: 1394845
URL: http://svn.apache.org/viewvc?rev=1394845&view=rev
Log:
Begin refactoring of JMS container code
Added:
camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsExchangeMessageHelper.java
- copied, changed from r1394668, camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsMessageExchangeHelper.java
camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/AbstractMessageHandler.java
- copied, changed from r1394668, camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/DefaultMessageHandler.java
camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/DefaultJmsKeyFormatStrategy.java
- copied, changed from r1394668, camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/DefaultJmsKeyFormatStrategy.java
camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/IllegalHeaderException.java
- copied, changed from r1394668, camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/IllegalHeaderException.java
camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/KeyFormatStrategy.java
- copied, changed from r1394668, camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/KeyFormatStrategy.java
Removed:
camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/DefaultJmsKeyFormatStrategy.java
camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/IllegalHeaderException.java
camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/KeyFormatStrategy.java
camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/DefaultConsumer.java
camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/DefaultMessageHandler.java
camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsMessageExchangeHelper.java
Modified:
camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsComponent.java
camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsConsumer.java
camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java
camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/InOnlyMessageHandler.java
camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/InOutMessageHandler.java
camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsMessageHelper.java
camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/producer/InOutProducer.java
Modified: camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsComponent.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsComponent.java?rev=1394845&r1=1394844&r2=1394845&view=diff
==============================================================================
--- camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsComponent.java (original)
+++ camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsComponent.java Fri Oct 5 22:36:20 2012
@@ -25,6 +25,7 @@ import org.apache.camel.Endpoint;
import org.apache.camel.ExchangePattern;
import org.apache.camel.component.sjms.jms.ConnectionFactoryResource;
import org.apache.camel.component.sjms.jms.ConnectionResource;
+import org.apache.camel.component.sjms.jms.KeyFormatStrategy;
import org.apache.camel.impl.DefaultComponent;
import org.apache.camel.spi.HeaderFilterStrategy;
import org.apache.camel.spi.HeaderFilterStrategyAware;
Modified: camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsConsumer.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsConsumer.java?rev=1394845&r1=1394844&r2=1394845&view=diff
==============================================================================
--- camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsConsumer.java (original)
+++ camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsConsumer.java Fri Oct 5 22:36:20 2012
@@ -16,33 +16,254 @@
*/
package org.apache.camel.component.sjms;
-import java.security.SecureRandom;
-import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+
+import javax.jms.Connection;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageListener;
+import javax.jms.Session;
import org.apache.camel.Endpoint;
+import org.apache.camel.ExchangePattern;
import org.apache.camel.Processor;
+import org.apache.camel.component.sjms.consumer.AbstractMessageHandler;
+import org.apache.camel.component.sjms.consumer.InOnlyMessageHandler;
+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.tx.BatchTransactionCommitStrategy;
+import org.apache.camel.component.sjms.tx.DefaultTransactionCommitStrategy;
+import org.apache.camel.component.sjms.tx.SessionTransactionSynchronization;
import org.apache.camel.impl.DefaultConsumer;
/**
- * TODO Add Class documentation for SjmsConsumer
+ * The SjmsConsumer is the base class for the SJMS MessageListener pool.
+ *
*/
public class SjmsConsumer extends DefaultConsumer {
+ protected MessageConsumerPool consumers;
+ private final ExecutorService executor;
+
+ /**
+ * A pool of MessageConsumerResources created at the initialization of the associated consumer.
+ */
+ protected class MessageConsumerPool extends ObjectPool<MessageConsumerResources> {
+
+ public MessageConsumerPool() {
+ super(getConsumerCount());
+ }
+
+ /**
+ * Creates a new MessageConsumerResources instance.
+ *
+ * @see org.apache.camel.component.sjms.jms.ObjectPool#createObject()
+ *
+ * @return
+ * @throws Exception
+ */
+ @Override
+ protected MessageConsumerResources createObject() throws Exception {
+ MessageConsumerResources model = null;
+ if (isTransacted() || getSjmsEndpoint().getExchangePattern().equals(ExchangePattern.InOut)) {
+ model = createConsumerWithDedicatedSession();
+ } else {
+ model = createConsumerListener();
+ }
+ return model;
+ }
+
+ /**
+ * Cleans up the MessageConsumerResources.
+ *
+ * @see org.apache.camel.component.sjms.jms.ObjectPool#destroyObject(java.lang.Object)
+ *
+ * @param model
+ * @throws Exception
+ */
+ @Override
+ protected void destroyObject(MessageConsumerResources model) throws Exception {
+ if (model != null) {
+ // First clean up our message consumer
+ if (model.getMessageConsumer() != null) {
+ model.getMessageConsumer().close();
+ }
+
+ // If the resource has a
+ if (model.getSession() != null) {
+ if (model.getSession().getTransacted()) {
+ try {
+ model.getSession().rollback();
+ } catch (Exception e) {
+ // Do nothing. Just make sure we are cleaned up
+ }
+ }
+ model.getSession().close();
+ }
+ }
+ }
+ }
+
+ protected class MessageConsumerResources {
+ private final Session session;
+ private final MessageConsumer messageConsumer;
+
+ /**
+ * @param messageProducer
+ */
+ public MessageConsumerResources(MessageConsumer messageConsumer) {
+ super();
+ this.session = null;
+ this.messageConsumer = messageConsumer;
+ }
+
+ /**
+ * @param session
+ * @param messageProducer
+ */
+ public MessageConsumerResources(Session session, MessageConsumer messageConsumer) {
+ super();
+ this.session = session;
+ this.messageConsumer = messageConsumer;
+ }
+
+ /**
+ * Gets the Session value of session for this instance of
+ * MessageProducerModel.
+ *
+ * @return the session
+ */
+ public Session getSession() {
+ return session;
+ }
+
+ /**
+ * Gets the QueueSender value of queueSender for this instance of
+ * MessageProducerModel.
+ *
+ * @return the queueSender
+ */
+ public MessageConsumer getMessageConsumer() {
+ return messageConsumer;
+ }
+ }
+
public SjmsConsumer(Endpoint endpoint, Processor processor) {
super(endpoint, processor);
+ this.executor = endpoint.getCamelContext().getExecutorServiceManager().newDefaultThreadPool(this, "SjmsConsumer");
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ consumers = new MessageConsumerPool();
+ consumers.fillPool();
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ super.doStop();
+ if (consumers != null) {
+ consumers.drainPool();
+ consumers = null;
+ }
+ }
+
+ @Override
+ protected void doResume() throws Exception {
+ super.doResume();
+ doStart();
+ }
+
+ @Override
+ protected void doSuspend() throws Exception {
+ doStop();
+ super.doSuspend();
}
/**
+ * Creates a {@link MessageConsumerResources} with a dedicated
+ * {@link Session} required for transacted and InOut consumers.
+ *
+ * @return MessageConsumerResources
+ * @throws Exception
+ */
+ 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);
+ }
+ MessageConsumer messageConsumer = JmsObjectFactory.createMessageConsumer(session, getDestinationName(), getMessageSelector(), isTopic(), getDurableSubscriptionId());
+ MessageListener handler = createMessageHandler(session);
+ messageConsumer.setMessageListener(handler);
+ getConnectionResource().returnConnection(conn);
+ return new MessageConsumerResources(session, messageConsumer);
+ }
+
+ /**
+ * Creates a {@link MessageConsumerResources} with a shared {@link Session}
+ * for non-transacted InOnly consumers.
+ *
* @return
+ * @throws Exception
*/
- protected String createId() {
- String answer = null;
- SecureRandom ng = new SecureRandom();
- UUID uuid = new UUID(ng.nextLong(), ng.nextLong());
- answer = uuid.toString();
- return answer;
+ 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
+ *
+ * @param session a session is only required if we are a transacted consumer
+ * @return
+ */
+ protected MessageListener createMessageHandler(Session session) {
+
+ TransactionCommitStrategy commitStrategy = null;
+ if (this.getTransactionCommitStrategy() != null) {
+ commitStrategy = this.getTransactionCommitStrategy();
+ } else if (this.getTransactionBatchCount() > 0) {
+ commitStrategy = new BatchTransactionCommitStrategy(this.getTransactionBatchCount());
+ } else {
+ commitStrategy = new DefaultTransactionCommitStrategy();
+ }
+
+ AbstractMessageHandler messageHandler = null;
+ if (getSjmsEndpoint().getExchangePattern().equals(ExchangePattern.InOnly)) {
+ if (isTransacted()) {
+ messageHandler = new InOnlyMessageHandler(getEndpoint(), executor, new SessionTransactionSynchronization(session, commitStrategy));
+ } else {
+ messageHandler = new InOnlyMessageHandler(getEndpoint(), executor);
+ }
+ } else {
+ if (isTransacted()) {
+ messageHandler = new InOutMessageHandler(getEndpoint(), executor, new SessionTransactionSynchronization(session, commitStrategy));
+ } else {
+ messageHandler = new InOutMessageHandler(getEndpoint(), executor);
+ }
+ }
+ messageHandler.setSession(session);
+ messageHandler.setProcessor(getAsyncProcessor());
+ messageHandler.setSynchronous(isSynchronous());
+ messageHandler.setTransacted(isTransacted());
+ messageHandler.setTopic(isTopic());
+ return messageHandler;
}
protected SjmsEndpoint getSjmsEndpoint() {
@@ -61,38 +282,85 @@ public class SjmsConsumer extends Defaul
return getSjmsEndpoint().getAcknowledgementMode().intValue();
}
+ /**
+ * Use to determine if transactions are enabled or disabled.
+ *
+ * @return true if transacted, otherwise false
+ */
public boolean isTransacted() {
return getSjmsEndpoint().isTransacted();
}
+ /**
+ * Use to determine whether or not to process exchanges synchronously.
+ *
+ * @return true if synchronous
+ */
public boolean isSynchronous() {
return getSjmsEndpoint().isSynchronous();
}
+ /**
+ * The destination name for this consumer.
+ *
+ * @return String
+ */
public String getDestinationName() {
return getSjmsEndpoint().getDestinationName();
}
+ /**
+ * Returns the number of consumer listeners.
+ *
+ * @return the consumerCount
+ */
public int getConsumerCount() {
return getSjmsEndpoint().getConsumerCount();
}
+ /**
+ * Flag set by the endpoint used by consumers and producers to determine if
+ * the consumer is a JMS Topic.
+ *
+ * @return the topic true if consumer is a JMS Topic, default is false
+ */
public boolean isTopic() {
return getSjmsEndpoint().isTopic();
}
+ /**
+ * Sets the JMS Message selector syntax.
+ *
+ * @param messageSelector Message selector syntax or null
+ */
public String getMessageSelector() {
return getSjmsEndpoint().getMessageSelector();
}
+ /**
+ * Gets the durable subscription Id.
+ *
+ * @return the durableSubscriptionId
+ */
public String getDurableSubscriptionId() {
return getSjmsEndpoint().getDurableSubscriptionId();
}
+ /**
+ * Gets the commit strategy.
+ *
+ * @return the transactionCommitStrategy
+ */
public TransactionCommitStrategy getTransactionCommitStrategy() {
return getSjmsEndpoint().getTransactionCommitStrategy();
}
+ /**
+ * If transacted, returns the nubmer of messages to be processed before
+ * committing the transaction.
+ *
+ * @return the transactionBatchCount
+ */
public int getTransactionBatchCount() {
return getSjmsEndpoint().getTransactionBatchCount();
}
Modified: camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java?rev=1394845&r1=1394844&r2=1394845&view=diff
==============================================================================
--- camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java (original)
+++ camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java Fri Oct 5 22:36:20 2012
@@ -23,8 +23,8 @@ import org.apache.camel.MultipleConsumer
import org.apache.camel.Processor;
import org.apache.camel.Producer;
import org.apache.camel.RuntimeCamelException;
-import org.apache.camel.component.sjms.consumer.DefaultConsumer;
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;
@@ -35,7 +35,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * TODO Add Class documentation for SjmsEndpoint
+ * A JMS Endpoint
*/
public class SjmsEndpoint extends DefaultEndpoint implements MultipleConsumersSupport {
protected final transient Logger logger = LoggerFactory.getLogger(getClass());
@@ -64,9 +64,9 @@ public class SjmsEndpoint extends Defaul
public SjmsEndpoint(String uri, Component component) {
super(uri, component);
if (getEndpointUri().indexOf("://queue:") > -1) {
- setTopic(false);
+ topic = false;
} else if (getEndpointUri().indexOf("://topic:") > -1) {
- setTopic(true);
+ topic = true;
} else {
throw new RuntimeCamelException("Endpoint URI unsupported: " + uri);
}
@@ -113,7 +113,7 @@ public class SjmsEndpoint extends Defaul
@Override
public Consumer createConsumer(Processor processor) throws Exception {
- return new DefaultConsumer(this, processor);
+ return new SjmsConsumer(this, processor);
}
@Override
@@ -150,131 +150,300 @@ public class SjmsEndpoint extends Defaul
return getSjmsComponent().getKeyFormatStrategy();
}
+ /**
+ * Returns a SessionPool if available.
+ *
+ * @return the sessions
+ */
public SessionPool getSessions() {
return sessions;
}
- public void setSynchronous(boolean asyncConsumer) {
- this.synchronous = asyncConsumer;
+ /**
+ * 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
+ */
public boolean isSynchronous() {
return synchronous;
}
- public void setTransacted(boolean transacted) {
- if (transacted) {
- setAcknowledgementMode(SessionAcknowledgementType.SESSION_TRANSACTED);
- }
- this.transacted = transacted;
- }
-
- public boolean isTransacted() {
- return transacted;
- }
-
- public void setNamedReplyTo(String namedReplyTo) {
- this.namedReplyTo = namedReplyTo;
- this.setExchangePattern(ExchangePattern.InOut);
- }
-
- public String getNamedReplyTo() {
- return namedReplyTo;
- }
-
- public void setAcknowledgementMode(SessionAcknowledgementType acknowledgementMode) {
- this.acknowledgementMode = acknowledgementMode;
+ /**
+ * Flag can be set to enable/disable synchronous exchange processing.
+ *
+ * @param synchronous true to process synchronously, default is true
+ */
+ public void setSynchronous(boolean synchronous) {
+ this.synchronous = synchronous;
}
+ /**
+ * Returns the configured acknowledgementMode.
+ *
+ * @return the acknowledgementMode
+ */
public SessionAcknowledgementType getAcknowledgementMode() {
return acknowledgementMode;
}
- public void setTopic(boolean topic) {
- this.topic = topic;
+ /**
+ * Sets the acknowledgementMode configured on this endpoint.
+ *
+ * @param acknowledgementMode default is
+ * SessionAcknowledgementType.AUTO_ACKNOWLEDGE
+ */
+ public void setAcknowledgementMode(SessionAcknowledgementType acknowledgementMode) {
+ this.acknowledgementMode = acknowledgementMode;
}
+ /**
+ * Flag set by the endpoint used by consumers and producers to determine if
+ * the endpoint is a JMS Topic.
+ *
+ * @return the topic true if endpoint is a JMS Topic, default is false
+ */
public boolean isTopic() {
return topic;
}
- public void setProducerCount(int producerCount) {
- this.producerCount = producerCount;
+ /**
+ * Returns the number of Session instances expected on this endpoint.
+ *
+ * @return the sessionCount
+ */
+ public int getSessionCount() {
+ return sessionCount;
+ }
+
+ /**
+ * Sets the number of Session instances used for this endpoint. Value is
+ * ignored for endpoints that require a dedicated session such as a
+ * transacted or InOut endpoint.
+ *
+ * @param sessionCount the number of Session instances, default is 1
+ */
+ public void setSessionCount(int sessionCount) {
+ this.sessionCount = sessionCount;
}
+ /**
+ * Returns the number of consumer listeners for this endpoint.
+ *
+ * @return the producerCount
+ */
public int getProducerCount() {
return producerCount;
}
- public void setConsumerCount(int consumerCount) {
- this.consumerCount = consumerCount;
+ /**
+ * Sets the number of producers used for this endpoint.
+ *
+ * @param producerCount the number of producers for this endpoint, default
+ * is 1
+ */
+ public void setProducerCount(int producerCount) {
+ this.producerCount = producerCount;
}
+ /**
+ * Returns the number of consumer listeners for this endpoint.
+ *
+ * @return the consumerCount
+ */
public int getConsumerCount() {
return consumerCount;
}
+ /**
+ * Sets the number of consumer listeners used for this endpoint.
+ *
+ * @param consumerCount the number of consumers for this endpoint, default
+ * is 1
+ */
+ public void setConsumerCount(int consumerCount) {
+ this.consumerCount = consumerCount;
+ }
+
+ /**
+ * Returns the Time To Live set on this endpoint.
+ *
+ * @return the ttl
+ */
public long getTtl() {
return ttl;
}
+ /**
+ * Flag used to adjust the Time To Live value of produced messages.
+ *
+ * @param ttl a new TTL, default is -1 (disabled)
+ */
public void setTtl(long ttl) {
this.ttl = ttl;
}
+ /**
+ * Use to determine if the enpoint has message persistence enabled or
+ * disabled.
+ *
+ * @return true if persistent, otherwise false
+ */
public boolean isPersistent() {
return persistent;
}
+ /**
+ * Flag used to enable/disable message persistence.
+ *
+ * @param persistent true if persistent, default is true
+ */
public void setPersistent(boolean persistent) {
this.persistent = persistent;
}
+ /**
+ * Gets the durable subscription Id.
+ *
+ * @return the durableSubscriptionId
+ */
public String getDurableSubscriptionId() {
return durableSubscriptionId;
}
- public void setDurableSubscriptionId(String durableSubscription) {
- this.durableSubscriptionId = durableSubscription;
+ /**
+ * Sets the durable subscription Id required for durable topics.
+ *
+ * @param durableSubscriptionId durable subscription Id or null
+ */
+ public void setDurableSubscriptionId(String durableSubscriptionId) {
+ this.durableSubscriptionId = durableSubscriptionId;
}
- public int getSessionCount() {
- return sessionCount;
- }
-
- public void setSessionCount(int sessionCount) {
- this.sessionCount = sessionCount;
+ /**
+ * Returns the InOut response timeout.
+ *
+ * @return the responseTimeOut
+ */
+ public long getResponseTimeOut() {
+ return responseTimeOut;
}
+ /**
+ * Sets the amount of time we should wait before timing out a InOut
+ * response.
+ *
+ * @param responseTimeOut response timeout
+ */
public void setResponseTimeOut(long responseTimeOut) {
this.responseTimeOut = responseTimeOut;
}
- public long getResponseTimeOut() {
- return responseTimeOut;
+ /**
+ * Returns the JMS Message selector syntax used to refine the messages being
+ * consumed.
+ *
+ * @return the messageSelector
+ */
+ public String getMessageSelector() {
+ return messageSelector;
}
+ /**
+ * Sets the JMS Message selector syntax.
+ *
+ * @param messageSelector Message selector syntax or null
+ */
public void setMessageSelector(String messageSelector) {
this.messageSelector = messageSelector;
}
- public String getMessageSelector() {
- return messageSelector;
+ /**
+ * If transacted, returns the nubmer of messages to be processed before
+ * committing the transaction.
+ *
+ * @return the transactionBatchCount
+ */
+ public int getTransactionBatchCount() {
+ return transactionBatchCount;
+ }
+
+ /**
+ * If transacted sets the number of messages to process before committing a
+ * transaction.
+ *
+ * @param transactionBatchCount number of messages to process before
+ * committing, default is 1
+ */
+ public void setTransactionBatchCount(int transactionBatchCount) {
+ this.transactionBatchCount = transactionBatchCount;
}
+ /**
+ * Gets the commit strategy.
+ *
+ * @return the transactionCommitStrategy
+ */
public TransactionCommitStrategy getTransactionCommitStrategy() {
return transactionCommitStrategy;
}
- public void setTransactionCommitStrategy(TransactionCommitStrategy commitStrategy) {
- this.transactionCommitStrategy = commitStrategy;
+ /**
+ * Sets the commit strategy.
+ *
+ * @param transactionCommitStrategy commit strategy to use when processing
+ * transacted messages
+ */
+ public void setTransactionCommitStrategy(TransactionCommitStrategy transactionCommitStrategy) {
+ this.transactionCommitStrategy = transactionCommitStrategy;
}
- public int getTransactionBatchCount() {
- return transactionBatchCount;
+ /**
+ * Use to determine if transactions are enabled or disabled.
+ *
+ * @return true if transacted, otherwise false
+ */
+ public boolean isTransacted() {
+ return transacted;
}
- public void setTransactionBatchCount(int transactionBatchCount) {
- this.transactionBatchCount = transactionBatchCount;
+ /**
+ * Enable/disable flag for transactions
+ *
+ * @param transacted true if transacted, otherwise false
+ */
+ public void setTransacted(boolean transacted) {
+ if (transacted) {
+ setAcknowledgementMode(SessionAcknowledgementType.SESSION_TRANSACTED);
+ }
+ this.transacted = transacted;
+ }
+
+ /**
+ * Returns the reply to destination name used for InOut producer endpoints.
+ *
+ * @return the namedReplyTo
+ */
+ public String getNamedReplyTo() {
+ return namedReplyTo;
+ }
+
+ /**
+ * Sets the reply to destination name used for InOut producer endpoints.
+ *
+ * @param the namedReplyTo the JMS reply to destination name
+ */
+ public void setNamedReplyTo(String namedReplyTo) {
+ this.namedReplyTo = namedReplyTo;
+ this.setExchangePattern(ExchangePattern.InOut);
}
}
Copied: camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsExchangeMessageHelper.java (from r1394668, camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsMessageExchangeHelper.java)
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsExchangeMessageHelper.java?p2=camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsExchangeMessageHelper.java&p1=camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsMessageExchangeHelper.java&r1=1394668&r2=1394845&rev=1394845&view=diff
==============================================================================
--- camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsMessageExchangeHelper.java (original)
+++ camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsExchangeMessageHelper.java Fri Oct 5 22:36:20 2012
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.camel.component.sjms.jms;
+package org.apache.camel.component.sjms;
import java.io.Serializable;
import java.util.Collection;
@@ -38,12 +38,14 @@ import javax.jms.TextMessage;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.RuntimeCamelException;
-import org.apache.camel.component.sjms.DefaultJmsKeyFormatStrategy;
-import org.apache.camel.component.sjms.IllegalHeaderException;
+import org.apache.camel.component.sjms.jms.DefaultJmsKeyFormatStrategy;
+import org.apache.camel.component.sjms.jms.IllegalHeaderException;
+import org.apache.camel.component.sjms.jms.JmsConstants;
+import org.apache.camel.component.sjms.jms.JmsMessageHeaderType;
+import org.apache.camel.component.sjms.jms.JmsMessageType;
import org.apache.camel.impl.DefaultMessage;
import org.apache.camel.util.ExchangeHelper;
import org.apache.camel.util.ObjectHelper;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -53,50 +55,49 @@ import static org.apache.camel.component
import static org.apache.camel.util.ObjectHelper.removeStartingCharacters;
/**
- * Utility class for {@link javax.jms.Message}.
- *
- * @version
+ * TODO Add Class documentation for SjmsExchangeMessageHelper
+ *
+ * @author sully6768
*/
-public final class JmsMessageExchangeHelper {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(JmsMessageExchangeHelper.class);
+public final class SjmsExchangeMessageHelper {
+ private static final Logger LOGGER = LoggerFactory.getLogger(SjmsExchangeMessageHelper.class);
- private JmsMessageExchangeHelper() {
+ private SjmsExchangeMessageHelper() {
}
-
+
public static Exchange createExchange(Message message, Endpoint endpoint) {
Exchange exchange = endpoint.createExchange();
return populateExchange(message, exchange, false);
}
-
+
@SuppressWarnings("unchecked")
public static Exchange populateExchange(Message message, Exchange exchange, boolean out) {
try {
- JmsMessageExchangeHelper.setJmsMessageHeaders(message, exchange, out);
+ SjmsExchangeMessageHelper.setJmsMessageHeaders(message, exchange, out);
if (message != null) {
// convert to JMS Message of the given type
DefaultMessage bodyMessage = null;
if (out) {
- bodyMessage = (DefaultMessage) exchange.getOut();
+ bodyMessage = (DefaultMessage)exchange.getOut();
} else {
- bodyMessage = (DefaultMessage) exchange.getIn();
+ bodyMessage = (DefaultMessage)exchange.getIn();
}
- switch (JmsMessageExchangeHelper.discoverType(message)) {
+ switch (SjmsExchangeMessageHelper.discoverType(message)) {
case Bytes:
- BytesMessage bytesMessage = (BytesMessage) message;
+ BytesMessage bytesMessage = (BytesMessage)message;
if (bytesMessage.getBodyLength() > Integer.MAX_VALUE) {
LOGGER.warn("Length of BytesMessage is too long: {}", bytesMessage.getBodyLength());
return null;
}
- byte[] result = new byte[(int) bytesMessage.getBodyLength()];
+ byte[] result = new byte[(int)bytesMessage.getBodyLength()];
bytesMessage.readBytes(result);
bodyMessage.setHeader(JMS_MESSAGE_TYPE, JmsMessageType.Bytes);
bodyMessage.setBody(result);
break;
case Map:
HashMap<String, Object> body = new HashMap<String, Object>();
- MapMessage mapMessage = (MapMessage) message;
+ MapMessage mapMessage = (MapMessage)message;
Enumeration<String> names = mapMessage.getMapNames();
while (names.hasMoreElements()) {
String key = names.nextElement();
@@ -107,12 +108,12 @@ public final class JmsMessageExchangeHel
bodyMessage.setBody(body);
break;
case Object:
- ObjectMessage objMsg = (ObjectMessage) message;
+ ObjectMessage objMsg = (ObjectMessage)message;
bodyMessage.setHeader(JMS_MESSAGE_TYPE, JmsMessageType.Object);
bodyMessage.setBody(objMsg.getObject());
break;
case Text:
- TextMessage textMsg = (TextMessage) message;
+ TextMessage textMsg = (TextMessage)message;
bodyMessage.setHeader(JMS_MESSAGE_TYPE, JmsMessageType.Text);
bodyMessage.setBody(textMsg.getText());
break;
@@ -131,9 +132,9 @@ public final class JmsMessageExchangeHel
/**
* Removes the property from the JMS message.
- *
+ *
* @param jmsMessage the JMS message
- * @param name name of the property to remove
+ * @param name name of the property to remove
* @return the old value of the property or <tt>null</tt> if not exists
* @throws JMSException can be thrown
*/
@@ -152,7 +153,7 @@ public final class JmsMessageExchangeHel
Map<String, Object> map = new LinkedHashMap<String, Object>();
Enumeration<?> en = jmsMessage.getPropertyNames();
while (en.hasMoreElements()) {
- String key = (String) en.nextElement();
+ String key = (String)en.nextElement();
if (name.equals(key)) {
answer = key;
} else {
@@ -171,16 +172,16 @@ public final class JmsMessageExchangeHel
/**
* Tests whether a given property with the name exists
- *
+ *
* @param jmsMessage the JMS message
- * @param name name of the property to test if exists
+ * @param name name of the property to test if exists
* @return <tt>true</tt> if the property exists, <tt>false</tt> if not.
* @throws JMSException can be thrown
*/
public static boolean hasProperty(Message jmsMessage, String name) throws JMSException {
Enumeration<?> en = jmsMessage.getPropertyNames();
while (en.hasMoreElements()) {
- String key = (String) en.nextElement();
+ String key = (String)en.nextElement();
if (name.equals(key)) {
return true;
}
@@ -190,10 +191,10 @@ public final class JmsMessageExchangeHel
/**
* Sets the property on the given JMS message.
- *
- * @param jmsMessage the JMS message
- * @param name name of the property to set
- * @param value the value
+ *
+ * @param jmsMessage the JMS message
+ * @param name name of the property to set
+ * @param value the value
* @throws JMSException can be thrown
*/
public static void setProperty(Message jmsMessage, String name, Object value) throws JMSException {
@@ -201,21 +202,21 @@ public final class JmsMessageExchangeHel
return;
}
if (value instanceof Byte) {
- jmsMessage.setByteProperty(name, (Byte) value);
+ jmsMessage.setByteProperty(name, (Byte)value);
} else if (value instanceof Boolean) {
- jmsMessage.setBooleanProperty(name, (Boolean) value);
+ jmsMessage.setBooleanProperty(name, (Boolean)value);
} else if (value instanceof Double) {
- jmsMessage.setDoubleProperty(name, (Double) value);
+ jmsMessage.setDoubleProperty(name, (Double)value);
} else if (value instanceof Float) {
- jmsMessage.setFloatProperty(name, (Float) value);
+ jmsMessage.setFloatProperty(name, (Float)value);
} else if (value instanceof Integer) {
- jmsMessage.setIntProperty(name, (Integer) value);
+ jmsMessage.setIntProperty(name, (Integer)value);
} else if (value instanceof Long) {
- jmsMessage.setLongProperty(name, (Long) value);
+ jmsMessage.setLongProperty(name, (Long)value);
} else if (value instanceof Short) {
- jmsMessage.setShortProperty(name, (Short) value);
+ jmsMessage.setShortProperty(name, (Short)value);
} else if (value instanceof String) {
- jmsMessage.setStringProperty(name, (String) value);
+ jmsMessage.setStringProperty(name, (String)value);
} else {
// fallback to Object
jmsMessage.setObjectProperty(name, value);
@@ -226,8 +227,8 @@ public final class JmsMessageExchangeHel
* Sets the correlation id on the JMS message.
* <p/>
* Will ignore exception thrown
- *
- * @param message the JMS message
+ *
+ * @param message the JMS message
* @param correlationId the correlation id
*/
public static void setCorrelationId(Message message, String correlationId) {
@@ -241,8 +242,9 @@ public final class JmsMessageExchangeHel
}
/**
- * Normalizes the destination name, by removing any leading queue or topic prefixes.
- *
+ * Normalizes the destination name, by removing any leading queue or topic
+ * prefixes.
+ *
* @param destination the destination
* @return the normalized destination
*/
@@ -261,9 +263,9 @@ public final class JmsMessageExchangeHel
/**
* Sets the JMSReplyTo on the message.
- *
- * @param message the message
- * @param replyTo the reply to destination
+ *
+ * @param message the message
+ * @param replyTo the reply to destination
*/
public static void setJMSReplyTo(Message message, Destination replyTo) {
try {
@@ -275,8 +277,8 @@ public final class JmsMessageExchangeHel
/**
* Gets the JMSReplyTo from the message.
- *
- * @param message the message
+ *
+ * @param message the message
* @return the reply to, can be <tt>null</tt>
*/
public static Destination getJMSReplyTo(Message message) {
@@ -291,8 +293,8 @@ public final class JmsMessageExchangeHel
/**
* Gets the JMSType from the message.
- *
- * @param message the message
+ *
+ * @param message the message
* @return the type, can be <tt>null</tt>
*/
public static String getJMSType(Message message) {
@@ -307,9 +309,10 @@ public final class JmsMessageExchangeHel
/**
* Gets the JMSRedelivered from the message.
- *
- * @param message the message
- * @return <tt>true</tt> if redelivered, <tt>false</tt> if not, <tt>null</tt> if not able to determine
+ *
+ * @param message the message
+ * @return <tt>true</tt> if redelivered, <tt>false</tt> if not,
+ * <tt>null</tt> if not able to determine
*/
public static Boolean getJMSRedelivered(Message message) {
try {
@@ -323,17 +326,18 @@ public final class JmsMessageExchangeHel
/**
* Sets the JMSDeliveryMode on the message.
- *
+ *
* @param exchange the exchange
- * @param message the message
- * @param deliveryMode the delivery mode, either as a String or integer
- * @throws javax.jms.JMSException is thrown if error setting the delivery mode
+ * @param message the message
+ * @param deliveryMode the delivery mode, either as a String or integer
+ * @throws javax.jms.JMSException is thrown if error setting the delivery
+ * mode
*/
public static void setJMSDeliveryMode(Exchange exchange, Message message, Object deliveryMode) throws JMSException {
Integer mode = null;
if (deliveryMode instanceof String) {
- String s = (String) deliveryMode;
+ String s = (String)deliveryMode;
if ("PERSISTENT".equalsIgnoreCase(s)) {
mode = DeliveryMode.PERSISTENT;
} else if ("NON_PERSISTENT".equalsIgnoreCase(s)) {
@@ -360,7 +364,6 @@ public final class JmsMessageExchangeHel
message.setIntProperty(JmsConstants.JMS_DELIVERY_MODE, mode);
}
}
-
public static Message setJmsMessageHeaders(final Exchange exchange, final Message jmsMessage) throws Exception {
Map<String, Object> headers = new HashMap<String, Object>(exchange.getIn().getHeaders());
@@ -372,7 +375,7 @@ public final class JmsMessageExchangeHel
} else if (headerName.equalsIgnoreCase("JMSReplyTo") && headerValue != null) {
if (headerValue instanceof String) {
// if the value is a String we must normalize it first
- headerValue = (String) headerValue;
+ headerValue = (String)headerValue;
} else {
// TODO write destination converter
// Destination replyTo =
@@ -385,7 +388,7 @@ public final class JmsMessageExchangeHel
} else if (headerName.equalsIgnoreCase("JMSPriority")) {
jmsMessage.setJMSPriority(ExchangeHelper.convertToType(exchange, Integer.class, headerValue));
} else if (headerName.equalsIgnoreCase("JMSDeliveryMode")) {
- JmsMessageExchangeHelper.setJMSDeliveryMode(exchange, jmsMessage, headerValue);
+ SjmsExchangeMessageHelper.setJMSDeliveryMode(exchange, jmsMessage, headerValue);
} else if (headerName.equalsIgnoreCase("JMSExpiration")) {
jmsMessage.setJMSExpiration(ExchangeHelper.convertToType(exchange, Long.class, headerValue));
} else {
@@ -395,9 +398,7 @@ public final class JmsMessageExchangeHel
// JMSMessageID, JMSTimestamp, JMSRedelivered
// log at trace level to not spam log
LOGGER.trace("Ignoring JMS header: {} with value: {}", headerName, headerValue);
- if (headerName.equalsIgnoreCase("JMSDestination")
- || headerName.equalsIgnoreCase("JMSMessageID")
- || headerName.equalsIgnoreCase("JMSTimestamp")
+ if (headerName.equalsIgnoreCase("JMSDestination") || headerName.equalsIgnoreCase("JMSMessageID") || headerName.equalsIgnoreCase("JMSTimestamp")
|| headerName.equalsIgnoreCase("JMSRedelivered")) {
// The following properties are set by the MessageProducer:
// JMSDestination
@@ -408,17 +409,17 @@ public final class JmsMessageExchangeHel
} else {
if (!(headerValue instanceof JmsMessageType)) {
String encodedName = new DefaultJmsKeyFormatStrategy().encodeKey(headerName);
- JmsMessageExchangeHelper.setProperty(jmsMessage, encodedName, headerValue);
+ SjmsExchangeMessageHelper.setProperty(jmsMessage, encodedName, headerValue);
}
}
- }
+ }
}
return jmsMessage;
}
-
+
public static JmsMessageType discoverType(Message value) throws Exception {
JmsMessageType answer = null;
- if (value != null) {
+ if (value != null) {
if (Message.class.isInstance(value)) {
if (BytesMessage.class.isInstance(value)) {
answer = JmsMessageType.Bytes;
@@ -435,9 +436,9 @@ public final class JmsMessageExchangeHel
}
return answer;
}
-
+
public static JmsMessageType discoverType(final Exchange exchange) {
- JmsMessageType answer = (JmsMessageType) exchange.getIn().getHeader(JMS_MESSAGE_TYPE);
+ JmsMessageType answer = (JmsMessageType)exchange.getIn().getHeader(JMS_MESSAGE_TYPE);
if (answer == null) {
final Object value = exchange.getIn().getBody();
if (value != null) {
@@ -458,53 +459,29 @@ public final class JmsMessageExchangeHel
}
@SuppressWarnings("unchecked")
- public static Exchange setJmsMessageHeaders(final Message jmsMessage, final Exchange exchange, boolean out)
- throws JMSException {
+ public static Exchange setJmsMessageHeaders(final Message jmsMessage, final Exchange exchange, boolean out) throws JMSException {
HashMap<String, Object> headers = new HashMap<String, Object>();
if (jmsMessage != null) {
// lets populate the standard JMS message headers
try {
- headers.put(
- JmsMessageHeaderType.JMSCorrelationID.toString(),
- jmsMessage.getJMSCorrelationID());
- headers.put(
- JmsMessageHeaderType.JMSDeliveryMode.toString(),
- jmsMessage.getJMSDeliveryMode());
- headers.put(
- JmsMessageHeaderType.JMSDestination.toString(),
- jmsMessage.getJMSDestination());
- headers.put(
- JmsMessageHeaderType.JMSExpiration.toString(),
- jmsMessage.getJMSExpiration());
- headers.put(
- JmsMessageHeaderType.JMSMessageID.toString(),
- jmsMessage.getJMSMessageID());
- headers.put(
- JmsMessageHeaderType.JMSPriority.toString(),
- jmsMessage.getJMSPriority());
- headers.put(
- JmsMessageHeaderType.JMSRedelivered.toString(),
- jmsMessage.getJMSRedelivered());
- headers.put(
- JmsMessageHeaderType.JMSTimestamp.toString(),
- jmsMessage.getJMSTimestamp());
- headers.put(
- JmsMessageHeaderType.JMSReplyTo.toString(),
- JmsMessageExchangeHelper.getJMSReplyTo(jmsMessage));
- headers.put(
- JmsMessageHeaderType.JMSType.toString(),
- JmsMessageExchangeHelper.getJMSType(jmsMessage));
+ headers.put(JmsMessageHeaderType.JMSCorrelationID.toString(), jmsMessage.getJMSCorrelationID());
+ headers.put(JmsMessageHeaderType.JMSDeliveryMode.toString(), jmsMessage.getJMSDeliveryMode());
+ headers.put(JmsMessageHeaderType.JMSDestination.toString(), jmsMessage.getJMSDestination());
+ headers.put(JmsMessageHeaderType.JMSExpiration.toString(), jmsMessage.getJMSExpiration());
+ headers.put(JmsMessageHeaderType.JMSMessageID.toString(), jmsMessage.getJMSMessageID());
+ headers.put(JmsMessageHeaderType.JMSPriority.toString(), jmsMessage.getJMSPriority());
+ headers.put(JmsMessageHeaderType.JMSRedelivered.toString(), jmsMessage.getJMSRedelivered());
+ headers.put(JmsMessageHeaderType.JMSTimestamp.toString(), jmsMessage.getJMSTimestamp());
+ headers.put(JmsMessageHeaderType.JMSReplyTo.toString(), SjmsExchangeMessageHelper.getJMSReplyTo(jmsMessage));
+ headers.put(JmsMessageHeaderType.JMSType.toString(), SjmsExchangeMessageHelper.getJMSType(jmsMessage));
// this works around a bug in the ActiveMQ property handling
- headers.put(
- JmsMessageHeaderType.JMSXGroupID.toString(),
- jmsMessage.getStringProperty(JmsMessageHeaderType.JMSXGroupID.toString()));
+ headers.put(JmsMessageHeaderType.JMSXGroupID.toString(), jmsMessage.getStringProperty(JmsMessageHeaderType.JMSXGroupID.toString()));
} catch (JMSException e) {
throw new RuntimeCamelException(e);
}
-
- for (Enumeration<String> enumeration = jmsMessage
- .getPropertyNames(); enumeration.hasMoreElements();) {
+
+ for (Enumeration<String> enumeration = jmsMessage.getPropertyNames(); enumeration.hasMoreElements();) {
String key = enumeration.nextElement();
if (hasIllegalHeaderKey(key)) {
throw new IllegalHeaderException("Header " + key + " is not a legal JMS header name value");
@@ -520,11 +497,11 @@ public final class JmsMessageExchangeHel
}
return exchange;
}
-
+
public static Message createMessage(Exchange exchange, Session session) throws Exception {
return createMessage(exchange, session, false);
}
-
+
@SuppressWarnings("unchecked")
public static Message createMessage(Exchange exchange, Session session, boolean out) throws Exception {
Message answer = null;
@@ -535,7 +512,7 @@ public final class JmsMessageExchangeHel
} else {
body = exchange.getIn().getBody();
}
- JmsMessageType messageType = JmsMessageExchangeHelper.discoverType(exchange);
+ JmsMessageType messageType = SjmsExchangeMessageHelper.discoverType(exchange);
switch (messageType) {
case Bytes:
@@ -572,61 +549,10 @@ public final class JmsMessageExchangeHel
throw e;
}
- answer = JmsMessageExchangeHelper.setJmsMessageHeaders(exchange, answer);
+ answer = SjmsExchangeMessageHelper.setJmsMessageHeaders(exchange, answer);
return answer;
}
-//
-// @SuppressWarnings("unchecked")
-// public static Message createMessage(Session session, Object payload, ) throws Exception {
-// Message answer = null;
-// Object body = null;
-// try {
-// if (out && exchange.getOut().getBody() != null) {
-// body = exchange.getOut().getBody();
-// } else {
-// body = exchange.getIn().getBody();
-// }
-// JmsMessageType messageType = JmsMessageExchangeHelper.discoverType(exchange);
-//
-// switch (messageType) {
-// case Bytes:
-// BytesMessage bytesMessage = session.createBytesMessage();
-// bytesMessage.writeBytes((byte[])body);
-// answer = bytesMessage;
-// break;
-// case Map:
-// MapMessage mapMessage = session.createMapMessage();
-// Map<String, Object> objMap = (Map<String, Object>)body;
-// Set<String> keys = objMap.keySet();
-// for (String key : keys) {
-// Object value = objMap.get(key);
-// mapMessage.setObject(key, value);
-// }
-// answer = mapMessage;
-// break;
-// case Object:
-// ObjectMessage objectMessage = session.createObjectMessage();
-// objectMessage.setObject((Serializable)body);
-// answer = objectMessage;
-// break;
-// case Text:
-// TextMessage textMessage = session.createTextMessage();
-// textMessage.setText((String)body);
-// answer = textMessage;
-// break;
-// default:
-// break;
-// }
-// } catch (Exception e) {
-// LOGGER.error("TODO Auto-generated catch block", e);
-// throw e;
-// }
-//
-// answer = JmsMessageExchangeHelper.setJmsMessageHeaders(exchange, answer);
-// return answer;
-// }
-//
-
+
private static boolean hasIllegalHeaderKey(String key) {
if (key == null) {
return true;
@@ -642,5 +568,4 @@ public final class JmsMessageExchangeHel
}
return false;
}
-
}
Copied: camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/AbstractMessageHandler.java (from r1394668, camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/DefaultMessageHandler.java)
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/AbstractMessageHandler.java?p2=camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/AbstractMessageHandler.java&p1=camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/DefaultMessageHandler.java&r1=1394668&r2=1394845&rev=1394845&view=diff
==============================================================================
--- camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/DefaultMessageHandler.java (original)
+++ camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/AbstractMessageHandler.java Fri Oct 5 22:36:20 2012
@@ -26,20 +26,19 @@ import org.apache.camel.AsyncProcessor;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.component.sjms.SjmsExchangeMessageHelper;
import org.apache.camel.component.sjms.TransactionCommitStrategy;
-import org.apache.camel.component.sjms.jms.JmsMessageExchangeHelper;
import org.apache.camel.impl.DefaultExchange;
import org.apache.camel.spi.Synchronization;
-import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.apache.camel.util.ObjectHelper.wrapRuntimeCamelException;
/**
- * TODO Add Class documentation for DefaultMessageHandler
+ * Abstract MessageListener
*/
-public abstract class DefaultMessageHandler implements MessageListener {
+public abstract class AbstractMessageHandler implements MessageListener {
protected final Logger log = LoggerFactory.getLogger(getClass());
@@ -54,30 +53,28 @@ public abstract class DefaultMessageHand
private boolean topic;
private TransactionCommitStrategy commitStrategy;
- public DefaultMessageHandler(Endpoint endpoint, ExecutorService executor) {
+ public AbstractMessageHandler(Endpoint endpoint, ExecutorService executor) {
this.endpoint = endpoint;
this.executor = executor;
}
- public DefaultMessageHandler(Endpoint endpoint, ExecutorService executor, Synchronization synchronization) {
+ public AbstractMessageHandler(Endpoint endpoint, ExecutorService executor, Synchronization synchronization) {
super();
this.synchronization = synchronization;
this.endpoint = endpoint;
this.executor = executor;
}
- @Override
- public void onMessage(Message message) {
- handleMessage(message);
- }
-
- /**
+ /*
+ * @see javax.jms.MessageListener#onMessage(javax.jms.Message)
+ *
* @param message
*/
- private void handleMessage(Message message) {
+ @Override
+ public void onMessage(Message message) {
RuntimeCamelException rce = null;
try {
- final DefaultExchange exchange = (DefaultExchange)JmsMessageExchangeHelper.createExchange(message, getEndpoint());
+ final DefaultExchange exchange = (DefaultExchange)SjmsExchangeMessageHelper.createExchange(message, getEndpoint());
log.debug("Processing Exchange.id:{}", exchange.getExchangeId());
@@ -87,16 +84,17 @@ public abstract class DefaultMessageHand
try {
if (isTransacted() || isSynchronous()) {
log.debug(" Handling synchronous message: {}", exchange.getIn().getBody());
- doHandleMessage(exchange);
+ handleMessage(exchange);
} else {
log.debug(" Handling asynchronous message: {}", exchange.getIn().getBody());
executor.execute(new Runnable() {
@Override
public void run() {
try {
- doHandleMessage(exchange);
+ handleMessage(exchange);
} catch (Exception e) {
- ObjectHelper.wrapRuntimeCamelException(e);
+ exchange.setException(e);
+// ObjectHelper.wrapRuntimeCamelException(e);
}
}
@@ -120,8 +118,14 @@ public abstract class DefaultMessageHand
}
}
- public abstract void doHandleMessage(final Exchange exchange);
+ /**
+ * @param exchange
+ */
+ public abstract void handleMessage(final Exchange exchange);
+ /**
+ * Method will be called to
+ */
public abstract void close();
public void setTransacted(boolean transacted) {
Modified: camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/InOnlyMessageHandler.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/InOnlyMessageHandler.java?rev=1394845&r1=1394844&r2=1394845&view=diff
==============================================================================
--- camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/InOnlyMessageHandler.java (original)
+++ camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/InOnlyMessageHandler.java Fri Oct 5 22:36:20 2012
@@ -25,10 +25,10 @@ import org.apache.camel.spi.Synchronizat
import org.apache.camel.util.AsyncProcessorHelper;
/**
- * An InOnly {@link DefaultMessageHandler}
+ * An InOnly {@link AbstractMessageHandler}
*
*/
-public class InOnlyMessageHandler extends DefaultMessageHandler {
+public class InOnlyMessageHandler extends AbstractMessageHandler {
/**
* @param endpoint
@@ -51,7 +51,7 @@ public class InOnlyMessageHandler extend
* @param message
*/
@Override
- public void doHandleMessage(final Exchange exchange) {
+ public void handleMessage(final Exchange exchange) {
if (log.isDebugEnabled()) {
log.debug("Handling InOnly Message: {}", exchange.getIn().getBody());
}
Modified: camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/InOutMessageHandler.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/InOutMessageHandler.java?rev=1394845&r1=1394844&r2=1394845&view=diff
==============================================================================
--- camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/InOutMessageHandler.java (original)
+++ camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/InOutMessageHandler.java Fri Oct 5 22:36:20 2012
@@ -33,19 +33,19 @@ import javax.jms.Topic;
import org.apache.camel.AsyncCallback;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
-import org.apache.camel.component.sjms.jms.JmsMessageExchangeHelper;
+import org.apache.camel.component.sjms.SjmsExchangeMessageHelper;
import org.apache.camel.component.sjms.jms.JmsObjectFactory;
import org.apache.camel.spi.Synchronization;
import org.apache.camel.util.AsyncProcessorHelper;
import org.apache.camel.util.ObjectHelper;
/**
- * TODO Add Class documentation for DefaultMessageHandler
+ * TODO Add Class documentation for AbstractMessageHandler
* TODO Create a producer
* cache manager to store and purge unused cashed producers or we will have a
* memory leak
*/
-public class InOutMessageHandler extends DefaultMessageHandler {
+public class InOutMessageHandler extends AbstractMessageHandler {
private Map<String, MessageProducer> producerCache = new TreeMap<String, MessageProducer>();
private ReadWriteLock lock = new ReentrantReadWriteLock();
@@ -74,7 +74,7 @@ public class InOutMessageHandler extends
* @param message
*/
@Override
- public void doHandleMessage(final Exchange exchange) {
+ public void handleMessage(final Exchange exchange) {
try {
MessageProducer messageProducer = null;
Object obj = exchange.getIn().getHeader("JMSReplyTo");
@@ -192,7 +192,7 @@ public class InOutMessageHandler extends
exchange.getOut().setBody(bytes);
}
}
- Message response = JmsMessageExchangeHelper.createMessage(exchange, getSession(), true);
+ Message response = SjmsExchangeMessageHelper.createMessage(exchange, getSession(), true);
response.setJMSCorrelationID(exchange.getIn().getHeader("JMSCorrelationID", String.class));
localProducer.send(response);
} catch (Exception e) {
Copied: camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/DefaultJmsKeyFormatStrategy.java (from r1394668, camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/DefaultJmsKeyFormatStrategy.java)
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/DefaultJmsKeyFormatStrategy.java?p2=camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/DefaultJmsKeyFormatStrategy.java&p1=camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/DefaultJmsKeyFormatStrategy.java&r1=1394668&r2=1394845&rev=1394845&view=diff
==============================================================================
--- camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/DefaultJmsKeyFormatStrategy.java (original)
+++ camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/DefaultJmsKeyFormatStrategy.java Fri Oct 5 22:36:20 2012
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.camel.component.sjms;
+package org.apache.camel.component.sjms.jms;
/**
* Default strategy that handles dots and hyphens.
Copied: camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/IllegalHeaderException.java (from r1394668, camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/IllegalHeaderException.java)
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/IllegalHeaderException.java?p2=camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/IllegalHeaderException.java&p1=camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/IllegalHeaderException.java&r1=1394668&r2=1394845&rev=1394845&view=diff
==============================================================================
--- camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/IllegalHeaderException.java (original)
+++ camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/IllegalHeaderException.java Fri Oct 5 22:36:20 2012
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.camel.component.sjms;
+package org.apache.camel.component.sjms.jms;
import org.apache.camel.RuntimeCamelException;
Modified: camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsMessageHelper.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsMessageHelper.java?rev=1394845&r1=1394844&r2=1394845&view=diff
==============================================================================
--- camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsMessageHelper.java (original)
+++ camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsMessageHelper.java Fri Oct 5 22:36:20 2012
@@ -32,9 +32,6 @@ import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.TextMessage;
-import org.apache.camel.component.sjms.DefaultJmsKeyFormatStrategy;
-import org.apache.camel.component.sjms.IllegalHeaderException;
-import org.apache.camel.component.sjms.KeyFormatStrategy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Copied: camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/KeyFormatStrategy.java (from r1394668, camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/KeyFormatStrategy.java)
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/KeyFormatStrategy.java?p2=camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/KeyFormatStrategy.java&p1=camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/KeyFormatStrategy.java&r1=1394668&r2=1394845&rev=1394845&view=diff
==============================================================================
--- camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/KeyFormatStrategy.java (original)
+++ camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/KeyFormatStrategy.java Fri Oct 5 22:36:20 2012
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.camel.component.sjms;
+package org.apache.camel.component.sjms.jms;
/**
* Strategy for applying encoding and decoding of JMS headers so they apply to
Modified: camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/producer/InOutProducer.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/producer/InOutProducer.java?rev=1394845&r1=1394844&r2=1394845&view=diff
==============================================================================
--- camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/producer/InOutProducer.java (original)
+++ camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/producer/InOutProducer.java Fri Oct 5 22:36:20 2012
@@ -38,8 +38,8 @@ import org.apache.camel.AsyncCallback;
import org.apache.camel.CamelException;
import org.apache.camel.Exchange;
import org.apache.camel.component.sjms.SjmsEndpoint;
+import org.apache.camel.component.sjms.SjmsExchangeMessageHelper;
import org.apache.camel.component.sjms.SjmsProducer;
-import org.apache.camel.component.sjms.jms.JmsMessageExchangeHelper;
import org.apache.camel.component.sjms.jms.JmsObjectFactory;
import org.apache.camel.component.sjms.jms.ObjectPool;
import org.apache.camel.component.sjms.tx.SessionTransactionSynchronization;
@@ -272,7 +272,7 @@ public class InOutProducer extends SjmsP
exchange.getUnitOfWork().addSynchronization(new SessionTransactionSynchronization(producer.getSession(), getCommitStrategy()));
}
- Message request = JmsMessageExchangeHelper.createMessage(exchange, producer.getSession());
+ Message request = SjmsExchangeMessageHelper.createMessage(exchange, producer.getSession());
// TODO just set the correlation id don't get it from the
// message
@@ -284,7 +284,7 @@ public class InOutProducer extends SjmsP
}
Object responseObject = null;
Exchanger<Object> messageExchanger = new Exchanger<Object>();
- JmsMessageExchangeHelper.setCorrelationId(request, correlationId);
+ SjmsExchangeMessageHelper.setCorrelationId(request, correlationId);
try {
lock.writeLock().lock();
exchangerMap.put(correlationId, messageExchanger);
@@ -293,7 +293,7 @@ public class InOutProducer extends SjmsP
}
MessageConsumerResource consumer = consumers.borrowObject(getResponseTimeOut());
- JmsMessageExchangeHelper.setJMSReplyTo(request, consumer.getReplyToDestination());
+ SjmsExchangeMessageHelper.setJMSReplyTo(request, consumer.getReplyToDestination());
consumers.returnObject(consumer);
producer.getMessageProducer().send(request);
@@ -327,7 +327,7 @@ public class InOutProducer extends SjmsP
exchange.setException((Throwable)responseObject);
} else if (responseObject instanceof Message) {
Message response = (Message)responseObject;
- JmsMessageExchangeHelper.populateExchange(response, exchange, true);
+ SjmsExchangeMessageHelper.populateExchange(response, exchange, true);
} else {
exchange.setException(new CamelException("Unknown response type: " + responseObject));
}