camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject [3/5] camel git commit: CAMEL-9116: camel-sjms should use same binding to/from JMS as camel-jms does.
Date Mon, 07 Sep 2015 09:38:52 GMT
CAMEL-9116: camel-sjms should use same binding to/from JMS as camel-jms does.


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/d19e5d74
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/d19e5d74
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/d19e5d74

Branch: refs/heads/master
Commit: d19e5d742b2cbed2f84ad2598bbe9c4789789d3b
Parents: 5b1d8da
Author: Claus Ibsen <davsclaus@apache.org>
Authored: Mon Sep 7 11:10:01 2015 +0200
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Mon Sep 7 11:10:01 2015 +0200

----------------------------------------------------------------------
 components/camel-sjms/pom.xml                   | 219 +++---
 .../camel/component/sjms/SjmsComponent.java     |  27 +-
 .../camel/component/sjms/SjmsConsumer.java      |   2 +-
 .../camel/component/sjms/SjmsEndpoint.java      | 127 ++-
 .../sjms/SjmsHeaderFilterStrategy.java          |  14 +-
 .../camel/component/sjms/SjmsMessage.java       | 283 +++++++
 .../component/sjms/batch/SjmsBatchConsumer.java |   5 +-
 .../component/sjms/batch/SjmsBatchEndpoint.java | 152 +++-
 .../sjms/consumer/AbstractMessageHandler.java   |  22 +-
 .../sjms/consumer/InOutMessageHandler.java      |   5 +-
 .../sjms/jms/DefaultJmsKeyFormatStrategy.java   |   2 +-
 .../camel/component/sjms/jms/JmsBinding.java    | 606 ++++++++++++++
 .../sjms/jms/JmsKeyFormatStrategy.java          |  41 +
 .../component/sjms/jms/JmsMessageHelper.java    | 782 ++++++-------------
 .../component/sjms/jms/KeyFormatStrategy.java   |  41 -
 .../sjms/jms/MessageCreatedStrategy.java        |  39 +
 .../component/sjms/producer/InOnlyProducer.java |  23 +-
 .../component/sjms/producer/InOutProducer.java  |  41 +-
 .../JMSMessageHelperTypeConversionTest.java     | 201 -----
 19 files changed, 1671 insertions(+), 961 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/d19e5d74/components/camel-sjms/pom.xml
----------------------------------------------------------------------
diff --git a/components/camel-sjms/pom.xml b/components/camel-sjms/pom.xml
index 6977796..3f8d023 100644
--- a/components/camel-sjms/pom.xml
+++ b/components/camel-sjms/pom.xml
@@ -15,121 +15,122 @@
     See the License for the specific language governing permissions and
     limitations under the License.
 -->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
-    <modelVersion>4.0.0</modelVersion>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+  <modelVersion>4.0.0</modelVersion>
 
-    <parent>
-        <groupId>org.apache.camel</groupId>
-        <artifactId>components</artifactId>
-        <version>2.16-SNAPSHOT</version>
-    </parent>
+  <parent>
+    <groupId>org.apache.camel</groupId>
+    <artifactId>components</artifactId>
+    <version>2.16-SNAPSHOT</version>
+  </parent>
 
-    <artifactId>camel-sjms</artifactId>
-    <packaging>bundle</packaging>
-    <name>Camel :: Simple JMS</name>
-    <description>A pure Java JMS Camel Component</description>
+  <artifactId>camel-sjms</artifactId>
+  <packaging>bundle</packaging>
+  <name>Camel :: Simple JMS</name>
+  <description>A pure Java JMS Camel Component</description>
 
-    <properties>
-        <camel.osgi.export.pkg>
-            org.apache.camel.component.sjms,
-            org.apache.camel.component.sjms.jms,
-            org.apache.camel.component.sjms.batch
-        </camel.osgi.export.pkg>
-        <camel.osgi.private.pkg>
-            org.apache.camel.component.sjms.consumer,
-            org.apache.camel.component.sjms.producer,
-            org.apache.camel.component.sjms.taskmanager,
-            org.apache.camel.component.sjms.tx
-        </camel.osgi.private.pkg>
-        <camel.osgi.export.service>
-            org.apache.camel.spi.ComponentResolver;component=sjms,
-            org.apache.camel.spi.ComponentResolver;component=sjms-batch
-        </camel.osgi.export.service>
-    </properties>
+  <properties>
+    <camel.osgi.export.pkg>
+      org.apache.camel.component.sjms,
+      org.apache.camel.component.sjms.jms,
+      org.apache.camel.component.sjms.batch
+    </camel.osgi.export.pkg>
+    <camel.osgi.private.pkg>
+      org.apache.camel.component.sjms.consumer,
+      org.apache.camel.component.sjms.producer,
+      org.apache.camel.component.sjms.taskmanager,
+      org.apache.camel.component.sjms.tx
+    </camel.osgi.private.pkg>
+    <camel.osgi.export.service>
+      org.apache.camel.spi.ComponentResolver;component=sjms,
+      org.apache.camel.spi.ComponentResolver;component=sjms-batch
+    </camel.osgi.export.service>
+  </properties>
 
-    <dependencies>
-        <dependency>
-            <groupId>org.apache.camel</groupId>
-            <artifactId>camel-core</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>commons-pool</groupId>
-            <artifactId>commons-pool</artifactId>
-        </dependency>
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.camel</groupId>
+      <artifactId>camel-core</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>commons-pool</groupId>
+      <artifactId>commons-pool</artifactId>
+    </dependency>
 
-        <dependency>
-            <groupId>org.apache.geronimo.specs</groupId>
-            <artifactId>geronimo-jms_1.1_spec</artifactId>
-            <scope>provided</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.geronimo.specs</groupId>
-            <artifactId>geronimo-annotation_1.0_spec</artifactId>
-            <version>${geronimo-annotation-spec-version}</version>
-            <scope>provided</scope>
-        </dependency>
+    <dependency>
+      <groupId>org.apache.geronimo.specs</groupId>
+      <artifactId>geronimo-jms_1.1_spec</artifactId>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.geronimo.specs</groupId>
+      <artifactId>geronimo-annotation_1.0_spec</artifactId>
+      <version>${geronimo-annotation-spec-version}</version>
+      <scope>provided</scope>
+    </dependency>
 
-        <dependency>
-            <groupId>org.apache.camel</groupId>
-            <artifactId>camel-test</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.activemq</groupId>
-            <artifactId>activemq-broker</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.activemq</groupId>
-            <artifactId>activemq-kahadb-store</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.activemq</groupId>
-            <artifactId>activemq-pool</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>com.atomikos</groupId>
-            <artifactId>transactions-jta</artifactId>
-            <version>${atomikos-transactions-version}</version>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.slf4j</groupId>
-            <artifactId>slf4j-log4j12</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>commons-io</groupId>
-            <artifactId>commons-io</artifactId>
-            <version>${commons-io-version}</version>
-            <scope>test</scope>
-        </dependency>
-    </dependencies>
+    <dependency>
+      <groupId>org.apache.camel</groupId>
+      <artifactId>camel-test</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.activemq</groupId>
+      <artifactId>activemq-broker</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.activemq</groupId>
+      <artifactId>activemq-kahadb-store</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.activemq</groupId>
+      <artifactId>activemq-pool</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.atomikos</groupId>
+      <artifactId>transactions-jta</artifactId>
+      <version>${atomikos-transactions-version}</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-log4j12</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>commons-io</groupId>
+      <artifactId>commons-io</artifactId>
+      <version>${commons-io-version}</version>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
 
-    <build>
-        <plugins>
-            <plugin>
-                <artifactId>maven-surefire-plugin</artifactId>
-                <configuration>
-          	    <forkCount>1</forkCount>
-	  	    <reuseForks>false</reuseForks>
-                    <forkedProcessTimeoutInSeconds>6000</forkedProcessTimeoutInSeconds>
-                </configuration>
-            </plugin>
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-clean-plugin</artifactId>
-                <configuration>
-                    <filesets>
-                        <fileset>
-                            <directory>${basedir}/activemq-data</directory>
-                        </fileset>
-                    </filesets>
-                </configuration>
-            </plugin>
-        </plugins>
-    </build>
+  <build>
+    <plugins>
+      <plugin>
+        <artifactId>maven-surefire-plugin</artifactId>
+        <configuration>
+          <forkCount>1</forkCount>
+          <reuseForks>false</reuseForks>
+          <forkedProcessTimeoutInSeconds>6000</forkedProcessTimeoutInSeconds>
+        </configuration>
+      </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-clean-plugin</artifactId>
+        <configuration>
+          <filesets>
+            <fileset>
+              <directory>${basedir}/activemq-data</directory>
+            </fileset>
+          </filesets>
+        </configuration>
+      </plugin>
+    </plugins>
+  </build>
 
 </project>

http://git-wip-us.apache.org/repos/asf/camel/blob/d19e5d74/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsComponent.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsComponent.java
index 3433ec9..a7347c7 100644
--- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsComponent.java
+++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsComponent.java
@@ -27,7 +27,8 @@ import org.apache.camel.component.sjms.jms.ConnectionFactoryResource;
 import org.apache.camel.component.sjms.jms.ConnectionResource;
 import org.apache.camel.component.sjms.jms.DefaultJmsKeyFormatStrategy;
 import org.apache.camel.component.sjms.jms.DestinationCreationStrategy;
-import org.apache.camel.component.sjms.jms.KeyFormatStrategy;
+import org.apache.camel.component.sjms.jms.JmsKeyFormatStrategy;
+import org.apache.camel.component.sjms.jms.MessageCreatedStrategy;
 import org.apache.camel.component.sjms.taskmanager.TimedTaskManager;
 import org.apache.camel.impl.UriEndpointComponent;
 import org.apache.camel.spi.HeaderFilterStrategy;
@@ -44,12 +45,13 @@ public class SjmsComponent extends UriEndpointComponent implements HeaderFilterS
     private ConnectionFactory connectionFactory;
     private ConnectionResource connectionResource;
     private HeaderFilterStrategy headerFilterStrategy = new SjmsHeaderFilterStrategy();
-    private KeyFormatStrategy keyFormatStrategy = new DefaultJmsKeyFormatStrategy();
+    private JmsKeyFormatStrategy jmsKeyFormatStrategy = new DefaultJmsKeyFormatStrategy();
     private Integer connectionCount = 1;
     private TransactionCommitStrategy transactionCommitStrategy;
     private TimedTaskManager timedTaskManager;
     private DestinationCreationStrategy destinationCreationStrategy;
     private ExecutorService asyncStartStopExecutorService;
+    private MessageCreatedStrategy messageCreatedStrategy;
 
     public SjmsComponent() {
         super(SjmsEndpoint.class);
@@ -200,12 +202,12 @@ public class SjmsComponent extends UriEndpointComponent implements HeaderFilterS
      * You can provide your own implementation of the org.apache.camel.component.jms.JmsKeyFormatStrategy
      * and refer to it using the # notation.
      */
-    public void setKeyFormatStrategy(KeyFormatStrategy keyFormatStrategy) {
-        this.keyFormatStrategy = keyFormatStrategy;
+    public void setJmsKeyFormatStrategy(JmsKeyFormatStrategy jmsKeyFormatStrategy) {
+        this.jmsKeyFormatStrategy = jmsKeyFormatStrategy;
     }
 
-    public KeyFormatStrategy getKeyFormatStrategy() {
-        return keyFormatStrategy;
+    public JmsKeyFormatStrategy getJmsKeyFormatStrategy() {
+        return jmsKeyFormatStrategy;
     }
 
     public TransactionCommitStrategy getTransactionCommitStrategy() {
@@ -241,4 +243,17 @@ public class SjmsComponent extends UriEndpointComponent implements HeaderFilterS
     public void setTimedTaskManager(TimedTaskManager timedTaskManager) {
         this.timedTaskManager = timedTaskManager;
     }
+
+    public MessageCreatedStrategy getMessageCreatedStrategy() {
+        return messageCreatedStrategy;
+    }
+
+    /**
+     * To use the given MessageCreatedStrategy which are invoked when Camel creates new instances of <tt>javax.jms.Message</tt>
+     * objects when Camel is sending a JMS message.
+     */
+    public void setMessageCreatedStrategy(MessageCreatedStrategy messageCreatedStrategy) {
+        this.messageCreatedStrategy = messageCreatedStrategy;
+    }
+
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/d19e5d74/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 322ad5c..162e252 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
@@ -206,7 +206,7 @@ public class SjmsConsumer extends DefaultConsumer {
      */
     protected MessageListener createMessageHandler(Session session) {
 
-        TransactionCommitStrategy commitStrategy = null;
+        TransactionCommitStrategy commitStrategy;
         if (getTransactionCommitStrategy() != null) {
             commitStrategy = getTransactionCommitStrategy();
         } else if (getTransactionBatchCount() > 0) {

http://git-wip-us.apache.org/repos/asf/camel/blob/d19e5d74/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 c7ed9ac..cb5d396 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
@@ -16,22 +16,30 @@
  */
 package org.apache.camel.component.sjms;
 
+import javax.jms.Message;
+import javax.jms.Session;
+
 import org.apache.camel.Component;
 import org.apache.camel.Consumer;
+import org.apache.camel.Exchange;
 import org.apache.camel.ExchangePattern;
 import org.apache.camel.MultipleConsumersSupport;
 import org.apache.camel.Processor;
 import org.apache.camel.Producer;
 import org.apache.camel.component.sjms.jms.ConnectionResource;
 import org.apache.camel.component.sjms.jms.DefaultDestinationCreationStrategy;
+import org.apache.camel.component.sjms.jms.DefaultJmsKeyFormatStrategy;
 import org.apache.camel.component.sjms.jms.DestinationCreationStrategy;
 import org.apache.camel.component.sjms.jms.DestinationNameParser;
-import org.apache.camel.component.sjms.jms.KeyFormatStrategy;
+import org.apache.camel.component.sjms.jms.JmsBinding;
+import org.apache.camel.component.sjms.jms.JmsKeyFormatStrategy;
+import org.apache.camel.component.sjms.jms.MessageCreatedStrategy;
 import org.apache.camel.component.sjms.jms.SessionAcknowledgementType;
 import org.apache.camel.component.sjms.producer.InOnlyProducer;
 import org.apache.camel.component.sjms.producer.InOutProducer;
 import org.apache.camel.impl.DefaultEndpoint;
 import org.apache.camel.spi.HeaderFilterStrategy;
+import org.apache.camel.spi.HeaderFilterStrategyAware;
 import org.apache.camel.spi.Metadata;
 import org.apache.camel.spi.UriEndpoint;
 import org.apache.camel.spi.UriParam;
@@ -43,11 +51,13 @@ import org.slf4j.LoggerFactory;
  * A JMS Endpoint
  */
 @UriEndpoint(scheme = "sjms", title = "Simple JMS", syntax = "sjms:destinationType:destinationName", consumerClass = SjmsConsumer.class, label = "messaging")
-public class SjmsEndpoint extends DefaultEndpoint implements MultipleConsumersSupport {
+public class SjmsEndpoint extends DefaultEndpoint implements MultipleConsumersSupport, HeaderFilterStrategyAware {
     protected final Logger logger = LoggerFactory.getLogger(getClass());
 
     private boolean topic;
 
+    private JmsBinding binding;
+
     @UriPath(enums = "queue,topic", defaultValue = "queue", description = "The kind of destination to use")
     private String destinationType;
     @UriPath @Metadata(required = "true")
@@ -55,6 +65,10 @@ public class SjmsEndpoint extends DefaultEndpoint implements MultipleConsumersSu
     @UriParam(label = "consumer", defaultValue = "true")
     private boolean synchronous = true;
     @UriParam
+    private HeaderFilterStrategy headerFilterStrategy;
+    @UriParam
+    private boolean includeAllJMSXProperties;
+    @UriParam
     private boolean transacted;
     @UriParam(label = "producer")
     private String namedReplyTo;
@@ -88,10 +102,16 @@ public class SjmsEndpoint extends DefaultEndpoint implements MultipleConsumersSu
     private boolean prefillPool = true;
     @UriParam(label = "producer", defaultValue = "true")
     private boolean allowNullBody = true;
+    @UriParam(defaultValue = "true")
+    private boolean mapJmsMessage = true;
     @UriParam
     private TransactionCommitStrategy transactionCommitStrategy;
     @UriParam
     private DestinationCreationStrategy destinationCreationStrategy = new DefaultDestinationCreationStrategy();
+    @UriParam
+    private MessageCreatedStrategy messageCreatedStrategy;
+    @UriParam
+    private JmsKeyFormatStrategy jmsKeyFormatStrategy;
 
     public SjmsEndpoint() {
     }
@@ -146,6 +166,34 @@ public class SjmsEndpoint extends DefaultEndpoint implements MultipleConsumersSu
         return true;
     }
 
+    public Exchange createExchange(Message message, Session session) {
+        Exchange exchange = createExchange(getExchangePattern());
+        exchange.setIn(new SjmsMessage(message, session, getBinding()));
+        return exchange;
+    }
+
+    public JmsBinding getBinding() {
+        if (binding == null) {
+            binding = createBinding();
+        }
+        return binding;
+    }
+
+    /**
+     * Creates the {@link org.apache.camel.component.sjms.jms.JmsBinding} to use.
+     */
+    protected JmsBinding createBinding() {
+        return new JmsBinding(isMapJmsMessage(), isAllowNullBody(), getHeaderFilterStrategy(), getJmsKeyFormatStrategy(), getMessageCreatedStrategy());
+    }
+
+    /**
+     * Sets the binding used to convert from a Camel message to and from a JMS
+     * message
+     */
+    public void setBinding(JmsBinding binding) {
+        this.binding = binding;
+    }
+
     /**
      * DestinationName is a JMS queue or topic name. By default, the destinationName is interpreted as a queue name.
      */
@@ -157,16 +205,35 @@ public class SjmsEndpoint extends DefaultEndpoint implements MultipleConsumersSu
         return destinationName;
     }
 
-    public ConnectionResource getConnectionResource() {
-        return getComponent().getConnectionResource();
+    public HeaderFilterStrategy getHeaderFilterStrategy() {
+        if (headerFilterStrategy == null) {
+            headerFilterStrategy = new SjmsHeaderFilterStrategy(isIncludeAllJMSXProperties());
+        }
+        return headerFilterStrategy;
+    }
+
+    /**
+     * To use a custom HeaderFilterStrategy to filter header to and from Camel message.
+     */
+    public void setHeaderFilterStrategy(HeaderFilterStrategy strategy) {
+        this.headerFilterStrategy = strategy;
     }
 
-    public HeaderFilterStrategy getSjmsHeaderFilterStrategy() {
-        return getComponent().getHeaderFilterStrategy();
+    public boolean isIncludeAllJMSXProperties() {
+        return includeAllJMSXProperties;
     }
 
-    public KeyFormatStrategy getJmsKeyFormatStrategy() {
-        return getComponent().getKeyFormatStrategy();
+    /**
+     * Whether to include all JMSXxxx properties when mapping from JMS to Camel Message.
+     * Setting this to true will include properties such as JMSXAppID, and JMSXUserID etc.
+     * Note: If you are using a custom headerFilterStrategy then this option does not apply.
+     */
+    public void setIncludeAllJMSXProperties(boolean includeAllJMSXProperties) {
+        this.includeAllJMSXProperties = includeAllJMSXProperties;
+    }
+
+    public ConnectionResource getConnectionResource() {
+        return getComponent().getConnectionResource();
     }
 
     public boolean isSynchronous() {
@@ -417,4 +484,48 @@ public class SjmsEndpoint extends DefaultEndpoint implements MultipleConsumersSu
     public void setAllowNullBody(boolean allowNullBody) {
         this.allowNullBody = allowNullBody;
     }
+
+    public boolean isMapJmsMessage() {
+        return mapJmsMessage;
+    }
+
+    /**
+     * Specifies whether Camel should auto map the received JMS message to a suited payload type, such as javax.jms.TextMessage to a String etc.
+     * See section about how mapping works below for more details.
+     */
+    public void setMapJmsMessage(boolean mapJmsMessage) {
+        this.mapJmsMessage = mapJmsMessage;
+    }
+
+    public MessageCreatedStrategy getMessageCreatedStrategy() {
+        return messageCreatedStrategy;
+    }
+
+    /**
+     * To use the given MessageCreatedStrategy which are invoked when Camel creates new instances of <tt>javax.jms.Message</tt>
+     * objects when Camel is sending a JMS message.
+     */
+    public void setMessageCreatedStrategy(MessageCreatedStrategy messageCreatedStrategy) {
+        this.messageCreatedStrategy = messageCreatedStrategy;
+    }
+
+    public JmsKeyFormatStrategy getJmsKeyFormatStrategy() {
+        if (jmsKeyFormatStrategy == null) {
+            jmsKeyFormatStrategy = new DefaultJmsKeyFormatStrategy();
+        }
+        return jmsKeyFormatStrategy;
+    }
+
+    /**
+     * Pluggable strategy for encoding and decoding JMS keys so they can be compliant with the JMS specification.
+     * Camel provides two implementations out of the box: default and passthrough.
+     * The default strategy will safely marshal dots and hyphens (. and -). The passthrough strategy leaves the key as is.
+     * Can be used for JMS brokers which do not care whether JMS header keys contain illegal characters.
+     * You can provide your own implementation of the org.apache.camel.component.jms.JmsKeyFormatStrategy
+     * and refer to it using the # notation.
+     */
+    public void setJmsKeyFormatStrategy(JmsKeyFormatStrategy jmsKeyFormatStrategy) {
+        this.jmsKeyFormatStrategy = jmsKeyFormatStrategy;
+    }
+
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/d19e5d74/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsHeaderFilterStrategy.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsHeaderFilterStrategy.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsHeaderFilterStrategy.java
index 82a8a90..0da77ca 100644
--- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsHeaderFilterStrategy.java
+++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsHeaderFilterStrategy.java
@@ -24,14 +24,18 @@ import org.apache.camel.impl.DefaultHeaderFilterStrategy;
 public class SjmsHeaderFilterStrategy extends DefaultHeaderFilterStrategy {
 
     public SjmsHeaderFilterStrategy() {
-        initialize();
+        this(false);
+    }
+
+    public SjmsHeaderFilterStrategy(boolean includeAllJMSXProperties) {
+        if (!includeAllJMSXProperties) {
+            initialize();
+        }
     }
 
     protected void initialize() {
-        // ignore provider specified JMS extension headers see page 39 of JMS
-        // 1.1 specification
-        // added "JMSXRecvTimestamp" as a workaround for an Oracle bug/typo in
-        // AqjmsMessage
+        // ignore provider specified JMS extension headers see page 39 of JMS 1.1 specification
+        // added "JMSXRecvTimestamp" as a workaround for an Oracle bug/typo in AqjmsMessage
         getOutFilter().add("JMSXUserID");
         getOutFilter().add("JMSXAppID");
         getOutFilter().add("JMSXDeliveryCount");

http://git-wip-us.apache.org/repos/asf/camel/blob/d19e5d74/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsMessage.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsMessage.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsMessage.java
new file mode 100644
index 0000000..92f8531
--- /dev/null
+++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsMessage.java
@@ -0,0 +1,283 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.sjms;
+
+import java.io.File;
+import java.util.Map;
+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;
+import org.apache.camel.component.sjms.jms.JmsBinding;
+import org.apache.camel.component.sjms.jms.JmsMessageHelper;
+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;
+
+/**
+ * Represents a {@link org.apache.camel.Message} for working with JMS
+ *
+ * @version
+ */
+public class SjmsMessage extends DefaultMessage {
+    private static final Logger LOG = LoggerFactory.getLogger(SjmsMessage.class);
+    private Message jmsMessage;
+    private Session jmsSession;
+    private JmsBinding binding;
+
+    public SjmsMessage(Message jmsMessage, Session jmsSession, JmsBinding binding) {
+        setJmsMessage(jmsMessage);
+        setJmsSession(jmsSession);
+        setBinding(binding);
+    }
+
+    @Override
+    public String toString() {
+        // do not print jmsMessage as there could be sensitive details
+        if (jmsMessage != null) {
+            try {
+                return "SjmsMessage[JmsMessageID: " + jmsMessage.getJMSMessageID() + "]";
+            } catch (Throwable e) {
+                // ignore
+            }
+        }
+        return "SjmsMessage@" + ObjectHelper.getIdentityHashCode(this);
+    }
+
+    @Override
+    public void copyFrom(org.apache.camel.Message that) {
+        if (that == this) {
+            // the same instance so do not need to copy
+            return;
+        }
+
+        // must initialize headers before we set the JmsMessage to avoid Camel
+        // populating it before we do the copy
+        getHeaders().clear();
+
+        boolean copyMessageId = true;
+        if (that instanceof SjmsMessage) {
+            SjmsMessage thatMessage = (SjmsMessage) that;
+            this.jmsMessage = thatMessage.jmsMessage;
+            if (this.jmsMessage != null) {
+                // for performance lets not copy the messageID if we are a JMS message
+                copyMessageId = false;
+            }
+        }
+        if (copyMessageId) {
+            setMessageId(that.getMessageId());
+        }
+
+        // copy body and fault flag
+        setBody(that.getBody());
+        setFault(that.isFault());
+
+        // we have already cleared the headers
+        if (that.hasHeaders()) {
+            getHeaders().putAll(that.getHeaders());
+        }
+
+        getAttachments().clear();
+        if (that.hasAttachments()) {
+            getAttachments().putAll(that.getAttachments());
+        }
+    }
+
+    public JmsBinding getBinding() {
+        if (binding == null) {
+            binding = ExchangeHelper.getBinding(getExchange(), JmsBinding.class);
+        }
+        return binding;
+    }
+
+    public void setBinding(JmsBinding binding) {
+        this.binding = binding;
+    }
+
+    /**
+     * Returns the underlying JMS message
+     */
+    public Message getJmsMessage() {
+        return jmsMessage;
+    }
+
+    public void setJmsMessage(Message jmsMessage) {
+        if (jmsMessage != null) {
+            try {
+                setMessageId(jmsMessage.getJMSMessageID());
+            } catch (JMSException e) {
+                LOG.warn("Unable to retrieve JMSMessageID from JMS Message", e);
+            }
+        }
+        this.jmsMessage = jmsMessage;
+    }
+
+    /**
+     * Returns the underlying JMS session.
+     * <p/>
+     * This may be <tt>null</tt>.
+     */
+    public Session getJmsSession() {
+        return jmsSession;
+    }
+
+    public void setJmsSession(Session jmsSession) {
+        this.jmsSession = jmsSession;
+    }
+
+    @Override
+    public void setBody(Object body) {
+        super.setBody(body);
+        if (body == null) {
+            // preserver headers even if we set body to null
+            ensureInitialHeaders();
+            // remove underlying jmsMessage since we mutated body to null
+            jmsMessage = null;
+        }
+    }
+
+    public Object getHeader(String name) {
+        Object answer = null;
+
+        // we will exclude using JMS-prefixed headers here to avoid strangeness with some JMS providers
+        // e.g. ActiveMQ returns the String not the Destination type for "JMSReplyTo"!
+        // only look in jms message directly if we have not populated headers
+        if (jmsMessage != null && !hasPopulatedHeaders() && !name.startsWith("JMS")) {
+            try {
+                // use binding to do the lookup as it has to consider using encoded keys
+                answer = getBinding().getObjectProperty(jmsMessage, name);
+            } catch (JMSException e) {
+                throw new RuntimeExchangeException("Unable to retrieve header from JMS Message: " + name, getExchange(), e);
+            }
+        }
+        // only look if we have populated headers otherwise there are no headers at all
+        // if we do lookup a header starting with JMS then force a lookup
+        if (answer == null && (hasPopulatedHeaders() || name.startsWith("JMS"))) {
+            answer = super.getHeader(name);
+        }
+        return answer;
+    }
+
+    @Override
+    public Map<String, Object> getHeaders() {
+        ensureInitialHeaders();
+        return super.getHeaders();
+    }
+
+    @Override
+    public Object removeHeader(String name) {
+        ensureInitialHeaders();
+        return super.removeHeader(name);
+    }
+
+    @Override
+    public void setHeaders(Map<String, Object> headers) {
+        ensureInitialHeaders();
+        super.setHeaders(headers);
+    }
+
+    @Override
+    public void setHeader(String name, Object value) {
+        ensureInitialHeaders();
+        super.setHeader(name, value);
+    }
+
+    @Override
+    public SjmsMessage newInstance() {
+        return new SjmsMessage(null, null, binding);
+    }
+
+    /**
+     * Returns true if a new JMS message instance should be created to send to the next component
+     */
+    public boolean shouldCreateNewMessage() {
+        return super.hasPopulatedHeaders();
+    }
+
+    /**
+     * Ensure that the headers have been populated from the underlying JMS message
+     * before we start mutating the headers
+     */
+    protected void ensureInitialHeaders() {
+        if (jmsMessage != null && !hasPopulatedHeaders()) {
+            // we have not populated headers so force this by creating
+            // new headers and set it on super
+            super.setHeaders(createHeaders());
+        }
+    }
+
+    @Override
+    protected Object createBody() {
+        if (jmsMessage != null) {
+            return getBinding().extractBodyFromJms(getExchange(), jmsMessage);
+        }
+        return null;
+    }
+
+    @Override
+    protected void populateInitialHeaders(Map<String, Object> map) {
+        if (jmsMessage != null && map != null) {
+            map.putAll(getBinding().extractHeadersFromJms(jmsMessage, getExchange()));
+        }
+    }
+
+    @Override
+    protected String createMessageId() {
+        if (jmsMessage == null) {
+            LOG.trace("No javax.jms.Message set so generating a new message id");
+            return super.createMessageId();
+        }
+        try {
+            String id = getDestinationAsString(jmsMessage.getJMSDestination()) + jmsMessage.getJMSMessageID();
+            return getSanitizedString(id);
+        } catch (JMSException e) {
+            throw new RuntimeExchangeException("Unable to retrieve JMSMessageID from JMS Message", getExchange(), e);
+        }
+    }
+
+    @Override
+    protected Boolean isTransactedRedelivered() {
+        if (jmsMessage != null) {
+            return JmsMessageHelper.getJMSRedelivered(jmsMessage);
+        } else {
+            return null;
+        }
+    }
+
+    private String getDestinationAsString(Destination destination) throws JMSException {
+        String result;
+        if (destination == null) {
+            result = "null destination!" + File.separator;
+        } else if (destination instanceof Topic) {
+            result = "topic" + File.separator + ((Topic) destination).getTopicName() + File.separator;
+        } else {
+            result = "queue" + File.separator + ((Queue) destination).getQueueName() + File.separator;
+        }
+        return result;
+    }
+
+    private String getSanitizedString(Object value) {
+        return value != null ? value.toString().replaceAll("[^a-zA-Z0-9\\.\\_\\-]", "_") : "";
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/d19e5d74/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchConsumer.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchConsumer.java
index ee2b250..e8ce161 100644
--- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchConsumer.java
+++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchConsumer.java
@@ -19,6 +19,7 @@ package org.apache.camel.component.sjms.batch;
 import java.io.PrintWriter;
 import java.io.StringWriter;
 import java.util.Date;
+import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -41,6 +42,7 @@ import org.apache.camel.Processor;
 import org.apache.camel.component.sjms.jms.JmsMessageHelper;
 import org.apache.camel.impl.DefaultConsumer;
 import org.apache.camel.processor.aggregate.AggregationStrategy;
+import org.apache.camel.util.ExchangeHelper;
 import org.apache.camel.util.ObjectHelper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -226,7 +228,8 @@ public class SjmsBatchConsumer extends DefaultConsumer {
                             LOG.debug("Message received: {}", messageCount);
                             if ((message instanceof ObjectMessage)
                                     || (message instanceof TextMessage)) {
-                                Exchange exchange = JmsMessageHelper.createExchange(message, getEndpoint());
+
+                                final Exchange exchange = getEndpoint().createExchange(message, session);
                                 aggregatedExchange = aggregationStrategy.aggregate(aggregatedExchange, exchange);
                                 aggregatedExchange.setProperty(SjmsBatchEndpoint.PROPERTY_BATCH_SIZE, messageCount);
                             } else {

http://git-wip-us.apache.org/repos/asf/camel/blob/d19e5d74/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchEndpoint.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchEndpoint.java
index b4c052f..49c74ba 100644
--- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchEndpoint.java
+++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchEndpoint.java
@@ -16,38 +16,65 @@
  */
 package org.apache.camel.component.sjms.batch;
 
+import javax.jms.Message;
+import javax.jms.Session;
+
 import org.apache.camel.Component;
 import org.apache.camel.Consumer;
+import org.apache.camel.Exchange;
 import org.apache.camel.Processor;
 import org.apache.camel.Producer;
+import org.apache.camel.component.sjms.SjmsHeaderFilterStrategy;
+import org.apache.camel.component.sjms.SjmsMessage;
+import org.apache.camel.component.sjms.jms.DefaultJmsKeyFormatStrategy;
 import org.apache.camel.component.sjms.jms.DestinationNameParser;
+import org.apache.camel.component.sjms.jms.JmsBinding;
+import org.apache.camel.component.sjms.jms.JmsKeyFormatStrategy;
+import org.apache.camel.component.sjms.jms.MessageCreatedStrategy;
 import org.apache.camel.impl.DefaultEndpoint;
 import org.apache.camel.processor.aggregate.AggregationStrategy;
+import org.apache.camel.spi.HeaderFilterStrategy;
+import org.apache.camel.spi.HeaderFilterStrategyAware;
 import org.apache.camel.spi.Metadata;
 import org.apache.camel.spi.UriEndpoint;
 import org.apache.camel.spi.UriParam;
 import org.apache.camel.spi.UriPath;
 
 @UriEndpoint(scheme = "sjms-batch", title = "Simple JMS Batch Component", syntax = "sjms-batch:destinationName",
-        consumerClass = SjmsBatchComponent.class, label = "messaging")
-public class SjmsBatchEndpoint extends DefaultEndpoint {
+        consumerClass = SjmsBatchComponent.class, label = "messaging", consumerOnly = true)
+public class SjmsBatchEndpoint extends DefaultEndpoint implements HeaderFilterStrategyAware {
 
     public static final int DEFAULT_COMPLETION_SIZE = 200; // the default dispatch queue size in ActiveMQ
     public static final int DEFAULT_COMPLETION_TIMEOUT = 500;
     public static final String PROPERTY_BATCH_SIZE = "CamelSjmsBatchSize";
 
-    @UriPath(label = "consumer") @Metadata(required = "true")
+    private JmsBinding binding;
+
+    @UriPath @Metadata(required = "true")
     private String destinationName;
-    @UriParam(label = "consumer", defaultValue = "1")
+    @UriParam(defaultValue = "1")
     private int consumerCount = 1;
-    @UriParam(label = "consumer", defaultValue = "200")
+    @UriParam(defaultValue = "200")
     private int completionSize = DEFAULT_COMPLETION_SIZE;
-    @UriParam(label = "consumer", defaultValue = "500")
+    @UriParam(defaultValue = "500")
     private int completionTimeout = DEFAULT_COMPLETION_TIMEOUT;
-    @UriParam(label = "consumer", defaultValue = "1000")
+    @UriParam(defaultValue = "1000")
     private int pollDuration = 1000;
-    @UriParam(label = "consumer") @Metadata(required = "true")
+    @UriParam @Metadata(required = "true")
     private AggregationStrategy aggregationStrategy;
+    @UriParam
+    private HeaderFilterStrategy headerFilterStrategy;
+    @UriParam
+    private boolean includeAllJMSXProperties;
+    @UriParam(defaultValue = "true")
+    private boolean allowNullBody = true;
+    @UriParam(defaultValue = "true")
+    private boolean mapJmsMessage = true;
+    @UriParam
+    private MessageCreatedStrategy messageCreatedStrategy;
+    @UriParam
+    private JmsKeyFormatStrategy jmsKeyFormatStrategy;
+
 
     public SjmsBatchEndpoint() {
     }
@@ -78,6 +105,34 @@ public class SjmsBatchEndpoint extends DefaultEndpoint {
         return new SjmsBatchConsumer(this, processor);
     }
 
+    public Exchange createExchange(Message message, Session session) {
+        Exchange exchange = createExchange(getExchangePattern());
+        exchange.setIn(new SjmsMessage(message, session, getBinding()));
+        return exchange;
+    }
+
+    public JmsBinding getBinding() {
+        if (binding == null) {
+            binding = createBinding();
+        }
+        return binding;
+    }
+
+    /**
+     * Creates the {@link org.apache.camel.component.sjms.jms.JmsBinding} to use.
+     */
+    protected JmsBinding createBinding() {
+        return new JmsBinding(isMapJmsMessage(), isAllowNullBody(), getHeaderFilterStrategy(), getJmsKeyFormatStrategy(), getMessageCreatedStrategy());
+    }
+
+    /**
+     * Sets the binding used to convert from a Camel message to and from a JMS
+     * message
+     */
+    public void setBinding(JmsBinding binding) {
+        this.binding = binding;
+    }
+
     public AggregationStrategy getAggregationStrategy() {
         return aggregationStrategy;
     }
@@ -141,4 +196,85 @@ public class SjmsBatchEndpoint extends DefaultEndpoint {
         this.pollDuration = pollDuration;
     }
 
+    public boolean isAllowNullBody() {
+        return allowNullBody;
+    }
+
+    /**
+     * Whether to allow sending messages with no body. If this option is false and the message body is null, then an JMSException is thrown.
+     */
+    public void setAllowNullBody(boolean allowNullBody) {
+        this.allowNullBody = allowNullBody;
+    }
+
+    public boolean isMapJmsMessage() {
+        return mapJmsMessage;
+    }
+
+    /**
+     * Specifies whether Camel should auto map the received JMS message to a suited payload type, such as javax.jms.TextMessage to a String etc.
+     * See section about how mapping works below for more details.
+     */
+    public void setMapJmsMessage(boolean mapJmsMessage) {
+        this.mapJmsMessage = mapJmsMessage;
+    }
+
+    public MessageCreatedStrategy getMessageCreatedStrategy() {
+        return messageCreatedStrategy;
+    }
+
+    /**
+     * To use the given MessageCreatedStrategy which are invoked when Camel creates new instances of <tt>javax.jms.Message</tt>
+     * objects when Camel is sending a JMS message.
+     */
+    public void setMessageCreatedStrategy(MessageCreatedStrategy messageCreatedStrategy) {
+        this.messageCreatedStrategy = messageCreatedStrategy;
+    }
+
+    public JmsKeyFormatStrategy getJmsKeyFormatStrategy() {
+        if (jmsKeyFormatStrategy == null) {
+            jmsKeyFormatStrategy = new DefaultJmsKeyFormatStrategy();
+        }
+        return jmsKeyFormatStrategy;
+    }
+
+    /**
+     * Pluggable strategy for encoding and decoding JMS keys so they can be compliant with the JMS specification.
+     * Camel provides two implementations out of the box: default and passthrough.
+     * The default strategy will safely marshal dots and hyphens (. and -). The passthrough strategy leaves the key as is.
+     * Can be used for JMS brokers which do not care whether JMS header keys contain illegal characters.
+     * You can provide your own implementation of the org.apache.camel.component.jms.JmsKeyFormatStrategy
+     * and refer to it using the # notation.
+     */
+    public void setJmsKeyFormatStrategy(JmsKeyFormatStrategy jmsKeyFormatStrategy) {
+        this.jmsKeyFormatStrategy = jmsKeyFormatStrategy;
+    }
+
+    public HeaderFilterStrategy getHeaderFilterStrategy() {
+        if (headerFilterStrategy == null) {
+            headerFilterStrategy = new SjmsHeaderFilterStrategy(isIncludeAllJMSXProperties());
+        }
+        return headerFilterStrategy;
+    }
+
+    /**
+     * To use a custom HeaderFilterStrategy to filter header to and from Camel message.
+     */
+    public void setHeaderFilterStrategy(HeaderFilterStrategy strategy) {
+        this.headerFilterStrategy = strategy;
+    }
+
+    public boolean isIncludeAllJMSXProperties() {
+        return includeAllJMSXProperties;
+    }
+
+    /**
+     * Whether to include all JMSXxxx properties when mapping from JMS to Camel Message.
+     * Setting this to true will include properties such as JMSXAppID, and JMSXUserID etc.
+     * Note: If you are using a custom headerFilterStrategy then this option does not apply.
+     */
+    public void setIncludeAllJMSXProperties(boolean includeAllJMSXProperties) {
+        this.includeAllJMSXProperties = includeAllJMSXProperties;
+    }
+
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/d19e5d74/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/AbstractMessageHandler.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/AbstractMessageHandler.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/AbstractMessageHandler.java
index 1598a43..f394008 100644
--- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/AbstractMessageHandler.java
+++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/AbstractMessageHandler.java
@@ -25,8 +25,6 @@ import org.apache.camel.AsyncProcessor;
 import org.apache.camel.Exchange;
 import org.apache.camel.RuntimeCamelException;
 import org.apache.camel.component.sjms.SjmsEndpoint;
-import org.apache.camel.component.sjms.jms.JmsMessageHelper;
-import org.apache.camel.impl.DefaultExchange;
 import org.apache.camel.spi.Synchronization;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -70,8 +68,7 @@ public abstract class AbstractMessageHandler implements MessageListener {
     public void onMessage(Message message) {
         RuntimeCamelException rce = null;
         try {
-            SjmsEndpoint endpoint = (SjmsEndpoint) getEndpoint();
-            final DefaultExchange exchange = (DefaultExchange) JmsMessageHelper.createExchange(message, endpoint, endpoint.getJmsKeyFormatStrategy());
+            final Exchange exchange = getEndpoint().createExchange(message, getSession());
 
             log.debug("Processing Exchange.id:{}", exchange.getExchangeId());
 
@@ -80,10 +77,10 @@ public abstract class AbstractMessageHandler implements MessageListener {
             }
             try {
                 if (isTransacted() || isSynchronous()) {
-                    log.debug("  Handling synchronous message: {}", exchange.getIn().getBody());
+                    log.debug("Handling synchronous message: {}", exchange.getIn().getBody());
                     handleMessage(exchange);
                 } else {
-                    log.debug("  Handling asynchronous message: {}", exchange.getIn().getBody());
+                    log.debug("Handling asynchronous message: {}", exchange.getIn().getBody());
                     executor.execute(new Runnable() {
                         @Override
                         public void run() {
@@ -96,12 +93,10 @@ public abstract class AbstractMessageHandler implements MessageListener {
                     });
                 }
             } catch (Exception e) {
-                if (exchange != null) {
-                    if (exchange.getException() == null) {
-                        exchange.setException(e);
-                    } else {
-                        throw e;
-                    }
+                if (exchange.getException() == null) {
+                    exchange.setException(e);
+                } else {
+                    throw e;
                 }
             }
         } catch (Exception e) {
@@ -113,9 +108,6 @@ public abstract class AbstractMessageHandler implements MessageListener {
         }
     }
 
-    /**
-     * @param exchange
-     */
     public abstract void handleMessage(final Exchange exchange);
 
     /**

http://git-wip-us.apache.org/repos/asf/camel/blob/d19e5d74/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/InOutMessageHandler.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/InOutMessageHandler.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/InOutMessageHandler.java
index 2068da5..97def12 100644
--- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/InOutMessageHandler.java
+++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/InOutMessageHandler.java
@@ -32,7 +32,6 @@ import org.apache.camel.AsyncCallback;
 import org.apache.camel.Exchange;
 import org.apache.camel.component.sjms.SjmsEndpoint;
 import org.apache.camel.component.sjms.jms.JmsConstants;
-import org.apache.camel.component.sjms.jms.JmsMessageHelper;
 import org.apache.camel.spi.Synchronization;
 
 /**
@@ -157,7 +156,9 @@ public class InOutMessageHandler extends AbstractMessageHandler {
         @Override
         public void done(boolean sync) {
             try {
-                Message response = JmsMessageHelper.createMessage(exchange, getSession(), getEndpoint());
+                // the response can either be in OUT or IN
+                org.apache.camel.Message msg = exchange.hasOut() ? exchange.getOut() : exchange.getIn();
+                Message response = getEndpoint().getBinding().makeJmsMessage(exchange, msg.getBody(), msg.getHeaders(), getSession(), null);
                 response.setJMSCorrelationID(exchange.getIn().getHeader(JmsConstants.JMS_CORRELATION_ID, String.class));
                 localProducer.send(response);
             } catch (Exception e) {

http://git-wip-us.apache.org/repos/asf/camel/blob/d19e5d74/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/DefaultJmsKeyFormatStrategy.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/DefaultJmsKeyFormatStrategy.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/DefaultJmsKeyFormatStrategy.java
index d95d2b5..4fa3308 100644
--- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/DefaultJmsKeyFormatStrategy.java
+++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/DefaultJmsKeyFormatStrategy.java
@@ -22,7 +22,7 @@ package org.apache.camel.component.sjms.jms;
  * This can be used for sending keys contain package names that is common by
  * Java frameworks.
  */
-public class DefaultJmsKeyFormatStrategy implements KeyFormatStrategy {
+public class DefaultJmsKeyFormatStrategy implements JmsKeyFormatStrategy {
 
     public String encodeKey(String key) {
         String answer = key.replace(".", "_DOT_");

http://git-wip-us.apache.org/repos/asf/camel/blob/d19e5d74/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsBinding.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsBinding.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsBinding.java
new file mode 100644
index 0000000..8dc2841
--- /dev/null
+++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsBinding.java
@@ -0,0 +1,606 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.sjms.jms;
+
+import java.io.File;
+import java.io.InputStream;
+import java.io.Reader;
+import java.io.Serializable;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.nio.ByteBuffer;
+import java.util.Date;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import javax.jms.BytesMessage;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.MapMessage;
+import javax.jms.Message;
+import javax.jms.MessageFormatException;
+import javax.jms.ObjectMessage;
+import javax.jms.Session;
+import javax.jms.StreamMessage;
+import javax.jms.TextMessage;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.NoTypeConversionAvailableException;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.StreamCache;
+import org.apache.camel.WrappedFile;
+import org.apache.camel.impl.DefaultExchangeHolder;
+import org.apache.camel.spi.HeaderFilterStrategy;
+import org.apache.camel.util.CamelContextHelper;
+import org.apache.camel.util.ExchangeHelper;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.w3c.dom.Node;
+
+import static org.apache.camel.component.sjms.jms.JmsMessageHelper.normalizeDestinationName;
+
+/**
+ * A Strategy used to convert between a Camel {@link org.apache.camel.Exchange} and {@link org.apache.camel.Message}
+ * to and from a JMS {@link javax.jms.Message}
+ */
+public class JmsBinding {
+
+    private static final Logger LOG = LoggerFactory.getLogger(JmsBinding.class);
+    private final boolean mapJmsMessage;
+    private final boolean allowNullBody;
+    private final HeaderFilterStrategy headerFilterStrategy;
+    private final JmsKeyFormatStrategy jmsJmsKeyFormatStrategy;
+    private final MessageCreatedStrategy messageCreatedStrategy;
+
+    public JmsBinding(boolean mapJmsMessage, boolean allowNullBody,
+                      HeaderFilterStrategy headerFilterStrategy, JmsKeyFormatStrategy jmsJmsKeyFormatStrategy,
+                      MessageCreatedStrategy messageCreatedStrategy) {
+        this.mapJmsMessage = mapJmsMessage;
+        this.allowNullBody = allowNullBody;
+        this.headerFilterStrategy = headerFilterStrategy;
+        this.jmsJmsKeyFormatStrategy = jmsJmsKeyFormatStrategy;
+        this.messageCreatedStrategy = messageCreatedStrategy;
+    }
+
+    /**
+     * Extracts the body from the JMS message
+     *
+     * @param exchange the exchange
+     * @param message  the message to extract its body
+     * @return the body, can be <tt>null</tt>
+     */
+    public Object extractBodyFromJms(Exchange exchange, Message message) {
+        try {
+
+            // TODO: new options to support
+
+            // is a custom message converter configured on endpoint then use it instead of doing the extraction
+            // based on message type
+/*            if (endpoint != null && endpoint.getMessageConverter() != null) {
+                if (LOG.isTraceEnabled()) {
+                    LOG.trace("Extracting body using a custom MessageConverter: {} from JMS message: {}", endpoint.getMessageConverter(), message);
+                }
+                return endpoint.getMessageConverter().fromMessage(message);
+            }
+*/
+            // if we are configured to not map the jms message then return it as body
+            if (!mapJmsMessage) {
+                LOG.trace("Option map JMS message is false so using JMS message as body: {}", message);
+                return message;
+            }
+
+            if (message instanceof ObjectMessage) {
+                LOG.trace("Extracting body as a ObjectMessage from JMS message: {}", message);
+                ObjectMessage objectMessage = (ObjectMessage)message;
+                Object payload = objectMessage.getObject();
+                if (payload instanceof DefaultExchangeHolder) {
+                    DefaultExchangeHolder holder = (DefaultExchangeHolder) payload;
+                    DefaultExchangeHolder.unmarshal(exchange, holder);
+                    return exchange.getIn().getBody();
+                } else {
+                    return objectMessage.getObject();
+                }
+            } else if (message instanceof TextMessage) {
+                LOG.trace("Extracting body as a TextMessage from JMS message: {}", message);
+                TextMessage textMessage = (TextMessage)message;
+                return textMessage.getText();
+            } else if (message instanceof MapMessage) {
+                LOG.trace("Extracting body as a MapMessage from JMS message: {}", message);
+                return createMapFromMapMessage((MapMessage)message);
+            } else if (message instanceof BytesMessage) {
+                LOG.trace("Extracting body as a BytesMessage from JMS message: {}", message);
+                return createByteArrayFromBytesMessage((BytesMessage)message);
+            } else if (message instanceof StreamMessage) {
+                LOG.trace("Extracting body as a StreamMessage from JMS message: {}", message);
+                return message;
+            } else {
+                return null;
+            }
+        } catch (JMSException e) {
+            throw new RuntimeCamelException("Failed to extract body due to: " + e + ". Message: " + message, e);
+        }
+    }
+
+    public Map<String, Object> extractHeadersFromJms(Message jmsMessage, Exchange exchange) {
+        Map<String, Object> map = new HashMap<String, Object>();
+        if (jmsMessage != null) {
+            // lets populate the standard JMS message headers
+            try {
+                map.put("JMSCorrelationID", jmsMessage.getJMSCorrelationID());
+                map.put("JMSCorrelationIDAsBytes", JmsMessageHelper.getJMSCorrelationIDAsBytes(jmsMessage));
+                map.put("JMSDeliveryMode", jmsMessage.getJMSDeliveryMode());
+                map.put("JMSDestination", jmsMessage.getJMSDestination());
+                map.put("JMSExpiration", jmsMessage.getJMSExpiration());
+                map.put("JMSMessageID", jmsMessage.getJMSMessageID());
+                map.put("JMSPriority", jmsMessage.getJMSPriority());
+                map.put("JMSRedelivered", jmsMessage.getJMSRedelivered());
+                map.put("JMSTimestamp", jmsMessage.getJMSTimestamp());
+
+                map.put("JMSReplyTo", JmsMessageHelper.getJMSReplyTo(jmsMessage));
+                map.put("JMSType", JmsMessageHelper.getJMSType(jmsMessage));
+
+                // this works around a bug in the ActiveMQ property handling
+                map.put(JmsConstants.JMSX_GROUP_ID, JmsMessageHelper.getStringProperty(jmsMessage, JmsConstants.JMSX_GROUP_ID));
+                map.put("JMSXUserID", JmsMessageHelper.getStringProperty(jmsMessage, "JMSXUserID"));
+            } catch (JMSException e) {
+                throw new RuntimeCamelException(e);
+            }
+
+            Enumeration<?> names;
+            try {
+                names = jmsMessage.getPropertyNames();
+            } catch (JMSException e) {
+                throw new RuntimeCamelException(e);
+            }
+            while (names.hasMoreElements()) {
+                String name = names.nextElement().toString();
+                try {
+                    Object value = JmsMessageHelper.getProperty(jmsMessage, name);
+                    if (headerFilterStrategy != null
+                            && headerFilterStrategy.applyFilterToExternalHeaders(name, value, exchange)) {
+                        continue;
+                    }
+
+                    // must decode back from safe JMS header name to original header name
+                    // when storing on this Camel JmsMessage object.
+                    String key = jmsJmsKeyFormatStrategy.decodeKey(name);
+                    map.put(key, value);
+                } catch (JMSException e) {
+                    throw new RuntimeCamelException(name, e);
+                }
+            }
+        }
+
+        return map;
+    }
+
+    public Object getObjectProperty(Message jmsMessage, String name) throws JMSException {
+        // try a direct lookup first
+        Object answer = jmsMessage.getObjectProperty(name);
+        if (answer == null) {
+            // then encode the key and do another lookup
+            String key = jmsJmsKeyFormatStrategy.encodeKey(name);
+            answer = jmsMessage.getObjectProperty(key);
+        }
+        return answer;
+    }
+
+    protected byte[] createByteArrayFromBytesMessage(BytesMessage message) throws JMSException {
+        if (message.getBodyLength() > Integer.MAX_VALUE) {
+            LOG.warn("Length of BytesMessage is too long: {}", message.getBodyLength());
+            return null;
+        }
+        byte[] result = new byte[(int)message.getBodyLength()];
+        message.readBytes(result);
+        return result;
+    }
+
+    /**
+     * Creates a JMS message from the Camel exchange and message
+     *
+     * @param exchange the current exchange
+     * @param session the JMS session used to create the message
+     * @return a newly created JMS Message instance containing the
+     * @throws JMSException if the message could not be created
+     */
+    public Message makeJmsMessage(Exchange exchange, Session session) throws JMSException {
+        Message answer = makeJmsMessage(exchange, exchange.getIn().getBody(), exchange.getIn().getHeaders(), session, null);
+        if (answer != null && messageCreatedStrategy != null) {
+            messageCreatedStrategy.onMessageCreated(answer, session, exchange, null);
+        }
+        return answer;
+    }
+
+    /**
+     * Creates a JMS message from the Camel exchange and message
+     *
+     * @param exchange the current exchange
+     * @param body the message body
+     * @param headers the message headers
+     * @param session the JMS session used to create the message
+     * @param cause optional exception occurred that should be sent as reply instead of a regular body
+     * @return a newly created JMS Message instance containing the
+     * @throws JMSException if the message could not be created
+     */
+    public Message makeJmsMessage(Exchange exchange, Object body, Map headers, Session session, Exception cause) throws JMSException {
+        Message answer = null;
+
+        // TODO: look at supporting some of these options
+
+/*        boolean alwaysCopy = endpoint != null && endpoint.getConfiguration().isAlwaysCopyMessage();
+        boolean force = endpoint != null && endpoint.getConfiguration().isForceSendOriginalMessage();
+        if (!alwaysCopy && camelMessage instanceof JmsMessage) {
+            JmsMessage jmsMessage = (JmsMessage)camelMessage;
+            if (!jmsMessage.shouldCreateNewMessage() || force) {
+                answer = jmsMessage.getJmsMessage();
+
+                if (!force) {
+                    // answer must match endpoint type
+                    JmsMessageType type = endpoint != null ? endpoint.getConfiguration().getJmsMessageType() : null;
+                    if (type != null && answer != null) {
+                        if (type == JmsMessageType.Text) {
+                            answer = answer instanceof TextMessage ? answer : null;
+                        } else if (type == JmsMessageType.Bytes) {
+                            answer = answer instanceof BytesMessage ? answer : null;
+                        } else if (type == JmsMessageType.Map) {
+                            answer = answer instanceof MapMessage ? answer : null;
+                        } else if (type == JmsMessageType.Object) {
+                            answer = answer instanceof ObjectMessage ? answer : null;
+                        } else if (type == JmsMessageType.Stream) {
+                            answer = answer instanceof StreamMessage ? answer : null;
+                        }
+                    }
+                }
+            }
+        }
+*/
+
+        if (answer == null) {
+            if (cause != null) {
+                // an exception occurred so send it as response
+                LOG.debug("Will create JmsMessage with caused exception: {}", cause);
+                // create jms message containing the caused exception
+                answer = createJmsMessage(cause, session);
+            } else {
+                // create regular jms message using the camel message body
+                answer = createJmsMessage(exchange, body, headers, session, exchange.getContext());
+                appendJmsProperties(answer, exchange, headers);
+            }
+        }
+
+        if (answer != null && messageCreatedStrategy != null) {
+            messageCreatedStrategy.onMessageCreated(answer, session, exchange, null);
+        }
+        return answer;
+    }
+
+    /**
+     * Appends the JMS headers from the Camel {@link Message}
+     */
+    public void appendJmsProperties(Message jmsMessage, Exchange exchange, Map<String, Object> headers) throws JMSException {
+        if (headers != null) {
+            Set<Map.Entry<String, Object>> entries = headers.entrySet();
+            for (Map.Entry<String, Object> entry : entries) {
+                String headerName = entry.getKey();
+                Object headerValue = entry.getValue();
+                appendJmsProperty(jmsMessage, exchange, headerName, headerValue);
+            }
+        }
+    }
+
+    public void appendJmsProperty(Message jmsMessage, Exchange exchange, String headerName, Object headerValue) throws JMSException {
+        if (isStandardJMSHeader(headerName)) {
+            if (headerName.equals("JMSCorrelationID")) {
+                jmsMessage.setJMSCorrelationID(ExchangeHelper.convertToType(exchange, String.class, headerValue));
+            } else if (headerName.equals("JMSReplyTo") && headerValue != null) {
+                if (headerValue instanceof String) {
+                    // if the value is a String we must normalize it first, and must include the prefix
+                    // as ActiveMQ requires that when converting the String to a javax.jms.Destination type
+                    headerValue = normalizeDestinationName((String) headerValue, true);
+                }
+                Destination replyTo = ExchangeHelper.convertToType(exchange, Destination.class, headerValue);
+                JmsMessageHelper.setJMSReplyTo(jmsMessage, replyTo);
+            } else if (headerName.equals("JMSType")) {
+                jmsMessage.setJMSType(ExchangeHelper.convertToType(exchange, String.class, headerValue));
+            } else if (headerName.equals("JMSPriority")) {
+                jmsMessage.setJMSPriority(ExchangeHelper.convertToType(exchange, Integer.class, headerValue));
+            } else if (headerName.equals("JMSDeliveryMode")) {
+                JmsMessageHelper.setJMSDeliveryMode(exchange, jmsMessage, headerValue);
+            } else if (headerName.equals("JMSExpiration")) {
+                jmsMessage.setJMSExpiration(ExchangeHelper.convertToType(exchange, Long.class, headerValue));
+            } else {
+                // The following properties are set by the MessageProducer:
+                // JMSDestination
+                // The following are set on the underlying JMS provider:
+                // JMSMessageID, JMSTimestamp, JMSRedelivered
+                // log at trace level to not spam log
+                LOG.trace("Ignoring JMS header: {} with value: {}", headerName, headerValue);
+            }
+        } else if (shouldOutputHeader(headerName, headerValue, exchange)) {
+            // only primitive headers and strings is allowed as properties
+            // see message properties: http://java.sun.com/j2ee/1.4/docs/api/javax/jms/Message.html
+            Object value = getValidJMSHeaderValue(headerName, headerValue);
+            if (value != null) {
+                // must encode to safe JMS header name before setting property on jmsMessage
+                String key = jmsJmsKeyFormatStrategy.encodeKey(headerName);
+                // set the property
+                JmsMessageHelper.setProperty(jmsMessage, key, value);
+            } else if (LOG.isDebugEnabled()) {
+                // okay the value is not a primitive or string so we cannot sent it over the wire
+                LOG.debug("Ignoring non primitive header: {} of class: {} with value: {}",
+                        new Object[]{headerName, headerValue.getClass().getName(), headerValue});
+            }
+        }
+    }
+
+    /**
+     * Is the given header a standard JMS header
+     * @param headerName the header name
+     * @return <tt>true</tt> if its a standard JMS header
+     */
+    protected boolean isStandardJMSHeader(String headerName) {
+        if (!headerName.startsWith("JMS")) {
+            return false;
+        }
+        if (headerName.startsWith("JMSX")) {
+            return false;
+        }
+        // vendors will use JMS_XXX as their special headers (where XXX is vendor name, such as JMS_IBM)
+        if (headerName.startsWith("JMS_")) {
+            return false;
+        }
+
+        // the 4th char must be a letter to be a standard JMS header
+        if (headerName.length() > 3) {
+            Character fourth = headerName.charAt(3);
+            if (Character.isLetter(fourth)) {
+                return true;
+            }
+        }
+
+        return false;
+    }
+
+    /**
+     * Strategy to test if the given header is valid according to the JMS spec to be set as a property
+     * on the JMS message.
+     * <p/>
+     * This default implementation will allow:
+     * <ul>
+     *   <li>any primitives and their counter Objects (Integer, Double etc.)</li>
+     *   <li>String and any other literals, Character, CharSequence</li>
+     *   <li>Boolean</li>
+     *   <li>Number</li>
+     *   <li>java.util.Date</li>
+     * </ul>
+     *
+     * @param headerName   the header name
+     * @param headerValue  the header value
+     * @return  the value to use, <tt>null</tt> to ignore this header
+     */
+    protected Object getValidJMSHeaderValue(String headerName, Object headerValue) {
+        if (headerValue instanceof String) {
+            return headerValue;
+        } else if (headerValue instanceof BigInteger) {
+            return headerValue.toString();
+        } else if (headerValue instanceof BigDecimal) {
+            return headerValue.toString();
+        } else if (headerValue instanceof Number) {
+            return headerValue;
+        } else if (headerValue instanceof Character) {
+            return headerValue;
+        } else if (headerValue instanceof CharSequence) {
+            return headerValue.toString();
+        } else if (headerValue instanceof Boolean) {
+            return headerValue;
+        } else if (headerValue instanceof Date) {
+            return headerValue.toString();
+        }
+        return null;
+    }
+
+    protected Message createJmsMessage(Exception cause, Session session) throws JMSException {
+        LOG.trace("Using JmsMessageType: {}", JmsMessageType.Object);
+        Message answer = session.createObjectMessage(cause);
+        // ensure default delivery mode is used by default
+        answer.setJMSDeliveryMode(Message.DEFAULT_DELIVERY_MODE);
+        return answer;
+    }
+
+    protected Message createJmsMessage(Exchange exchange, Object body, Map<String, Object> headers, Session session, CamelContext context) throws JMSException {
+        JmsMessageType type = null;
+
+        // TODO: support some of these options?
+
+/*        // special for transferExchange
+        if (endpoint != null && endpoint.isTransferExchange()) {
+            LOG.trace("Option transferExchange=true so we use JmsMessageType: Object");
+            Serializable holder = DefaultExchangeHolder.marshal(exchange);
+            Message answer = session.createObjectMessage(holder);
+            // ensure default delivery mode is used by default
+            answer.setJMSDeliveryMode(Message.DEFAULT_DELIVERY_MODE);
+            return answer;
+        }
+
+        // use a custom message converter
+        if (endpoint != null && endpoint.getMessageConverter() != null) {
+            if (LOG.isTraceEnabled()) {
+                LOG.trace("Creating JmsMessage using a custom MessageConverter: {} with body: {}", endpoint.getMessageConverter(), body);
+            }
+            return endpoint.getMessageConverter().toMessage(body, session);
+        }
+*/
+        // check if header have a type set, if so we force to use it
+/*
+        if (headers.containsKey(JmsConstants.JMS_MESSAGE_TYPE)) {
+            type = context.getTypeConverter().convertTo(JmsMessageType.class, headers.get(JmsConstants.JMS_MESSAGE_TYPE));
+        } else if (endpoint != null && endpoint.getConfiguration().getJmsMessageType() != null) {
+            // force a specific type from the endpoint configuration
+            type = endpoint.getConfiguration().getJmsMessageType();
+        } else {
+*/            type = getJMSMessageTypeForBody(exchange, body, headers, session, context);
+        //}
+
+        // create the JmsMessage based on the type
+        if (type != null) {
+            if (body == null && !allowNullBody) {
+                throw new JMSException("Cannot send message as message body is null, and option allowNullBody is false.");
+            }
+            LOG.trace("Using JmsMessageType: {}", type);
+            Message answer = createJmsMessageForType(exchange, body, headers, session, context, type);
+            // ensure default delivery mode is used by default
+            answer.setJMSDeliveryMode(Message.DEFAULT_DELIVERY_MODE);
+            return answer;
+        }
+
+        // check for null body
+        if (body == null && !allowNullBody) {
+            throw new JMSException("Cannot send message as message body is null, and option allowNullBody is false.");
+        }
+
+        // warn if the body could not be mapped
+        if (body != null && LOG.isWarnEnabled()) {
+            LOG.warn("Cannot determine specific JmsMessage type to use from body class."
+                    + " Will use generic JmsMessage."
+                    + " Body class: " + ObjectHelper.classCanonicalName(body)
+                    + ". If you want to send a POJO then your class might need to implement java.io.Serializable"
+                    + ", or you can force a specific type by setting the jmsMessageType option on the JMS endpoint.");
+        }
+
+        // return a default message
+        Message answer = session.createMessage();
+        // ensure default delivery mode is used by default
+        answer.setJMSDeliveryMode(Message.DEFAULT_DELIVERY_MODE);
+        return answer;
+    }
+
+    /**
+     * Return the {@link JmsMessageType}
+     *
+     * @return type or null if no mapping was possible
+     */
+    protected JmsMessageType getJMSMessageTypeForBody(Exchange exchange, Object body, Map<String, Object> headers, Session session, CamelContext context) {
+        JmsMessageType type = null;
+        // let body determine the type
+        if (body instanceof Node || body instanceof String) {
+            type = JmsMessageType.Text;
+        } else if (body instanceof byte[] || body instanceof WrappedFile || body instanceof File || body instanceof Reader
+                || body instanceof InputStream || body instanceof ByteBuffer || body instanceof StreamCache) {
+            type = JmsMessageType.Bytes;
+        } else if (body instanceof Map) {
+            type = JmsMessageType.Map;
+        } else if (body instanceof Serializable) {
+            type = JmsMessageType.Object;
+        } else if (exchange.getContext().getTypeConverter().tryConvertTo(File.class, body) != null
+                || exchange.getContext().getTypeConverter().tryConvertTo(InputStream.class, body) != null) {
+            type = JmsMessageType.Bytes;
+        }
+        return type;
+    }
+
+    /**
+     *
+     * Create the {@link Message}
+     *
+     * @return jmsMessage or null if the mapping was not successfully
+     */
+    protected Message createJmsMessageForType(Exchange exchange, Object body, Map<String, Object> headers, Session session, CamelContext context, JmsMessageType type) throws JMSException {
+        switch (type) {
+            case Text: {
+                TextMessage message = session.createTextMessage();
+                if (body != null) {
+                    String payload = context.getTypeConverter().convertTo(String.class, exchange, body);
+                    message.setText(payload);
+                }
+                return message;
+            }
+            case Bytes: {
+                BytesMessage message = session.createBytesMessage();
+                if (body != null) {
+                    byte[] payload = context.getTypeConverter().convertTo(byte[].class, exchange, body);
+                    message.writeBytes(payload);
+                }
+                return message;
+            }
+            case Map: {
+                MapMessage message = session.createMapMessage();
+                if (body != null) {
+                    Map<?, ?> payload = context.getTypeConverter().convertTo(Map.class, exchange, body);
+                    populateMapMessage(message, payload, context);
+                }
+                return message;
+            }
+            case Object:
+                ObjectMessage message = session.createObjectMessage();
+                if (body != null) {
+                    try {
+                        Serializable payload = context.getTypeConverter().mandatoryConvertTo(Serializable.class, exchange, body);
+                        message.setObject(payload);
+                    } catch (NoTypeConversionAvailableException e) {
+                        // cannot convert to serializable then thrown an exception to avoid sending a null message
+                        JMSException cause = new MessageFormatException(e.getMessage());
+                        cause.initCause(e);
+                        throw cause;
+                    }
+                }
+                return message;
+            default:
+                break;
+        }
+        return null;
+    }
+    /**
+     * Populates a {@link MapMessage} from a {@link Map} instance.
+     */
+    protected void populateMapMessage(MapMessage message, Map<?, ?> map, CamelContext context)
+            throws JMSException {
+        for (Map.Entry<?, ?> entry : map.entrySet()) {
+            String keyString = CamelContextHelper.convertTo(context, String.class, entry.getKey());
+            if (keyString != null) {
+                message.setObject(keyString, entry.getValue());
+            }
+        }
+    }
+
+    /**
+     * Extracts a {@link Map} from a {@link MapMessage}
+     */
+    public Map<String, Object> createMapFromMapMessage(MapMessage message) throws JMSException {
+        Map<String, Object> answer = new HashMap<String, Object>();
+        Enumeration<?> names = message.getMapNames();
+        while (names.hasMoreElements()) {
+            String name = names.nextElement().toString();
+            Object value = message.getObject(name);
+            answer.put(name, value);
+        }
+        return answer;
+    }
+
+    /**
+     * Strategy to allow filtering of headers which are put on the JMS message
+     * <p/>
+     * <b>Note</b>: Currently only supports sending java identifiers as keys
+     */
+    protected boolean shouldOutputHeader(String headerName, Object headerValue, Exchange exchange) {
+        return headerFilterStrategy == null
+                || !headerFilterStrategy.applyFilterToCamelHeaders(headerName, headerValue, exchange);
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/d19e5d74/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsKeyFormatStrategy.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsKeyFormatStrategy.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsKeyFormatStrategy.java
new file mode 100644
index 0000000..5a0327e
--- /dev/null
+++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsKeyFormatStrategy.java
@@ -0,0 +1,41 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.sjms.jms;
+
+/**
+ * Strategy for applying encoding and decoding of JMS headers so they apply to
+ * the JMS spec.
+ */
+public interface JmsKeyFormatStrategy {
+
+    /**
+     * Encodes the key before its sent as a {@link javax.jms.Message} message.
+     *
+     * @param key the original key
+     * @return the encoded key
+     */
+    String encodeKey(String key);
+
+    /**
+     * Decodes the key after its received from a {@link javax.jms.Message}
+     * message.
+     *
+     * @param key the encoded key
+     * @return the decoded key as the original key
+     */
+    String decodeKey(String key);
+}


Mime
View raw message