Repository: camel Updated Branches: refs/heads/master f10623e0e -> dd06fecc3
CAMEL-11739: camel-jms - Allow to configure a list of header names to preserve despite being invalid JMS spec type Project: http://git-wip-us.apache.org/repos/asf/camel/repo Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/39057c81 Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/39057c81 Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/39057c81 Branch: refs/heads/master Commit: 39057c8101b71e20c939f928b65f0ca37b5a1d0c Parents: f10623e Author: Claus Ibsen <davscl...@apache.org> Authored: Sat Sep 2 09:52:49 2017 +0200 Committer: Claus Ibsen <davscl...@apache.org> Committed: Sat Sep 2 09:52:49 2017 +0200 ---------------------------------------------------------------------- .../camel-jms/src/main/docs/jms-component.adoc | 6 +- .../apache/camel/component/jms/JmsBinding.java | 16 +++++ .../camel/component/jms/JmsComponent.java | 13 ++++ .../camel/component/jms/JmsConfiguration.java | 18 +++++ .../apache/camel/component/jms/JmsEndpoint.java | 10 +++ .../jms/JmsAllowAdditionalHeadersTest.java | 74 ++++++++++++++++++++ .../springboot/JmsComponentConfiguration.java | 33 +++++++++ 7 files changed, 168 insertions(+), 2 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/camel/blob/39057c81/components/camel-jms/src/main/docs/jms-component.adoc ---------------------------------------------------------------------- diff --git a/components/camel-jms/src/main/docs/jms-component.adoc b/components/camel-jms/src/main/docs/jms-component.adoc index 55ef547..f25b36e 100644 --- a/components/camel-jms/src/main/docs/jms-component.adoc +++ b/components/camel-jms/src/main/docs/jms-component.adoc @@ -199,7 +199,7 @@ about these properties by consulting the relevant Spring documentation. // component options: START -The JMS component supports 75 options which are listed below. +The JMS component supports 76 options which are listed below. @@ -274,6 +274,7 @@ The JMS component supports 75 options which are listed below. | **includeAllJMSX Properties** (advanced) | 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. | false | boolean | **defaultTaskExecutor Type** (consumer) | Specifies what default TaskExecutor type to use in the DefaultMessageListenerContainer for both consumer endpoints and the ReplyTo consumer of producer endpoints. Possible values: SimpleAsync (uses Spring's SimpleAsyncTaskExecutor) or ThreadPool (uses Spring's ThreadPoolTaskExecutor with optimal values - cached threadpool-like). If not set it defaults to the previous behaviour which uses a cached thread pool for consumer endpoints and SimpleAsync for reply consumers. The use of ThreadPool is recommended to reduce thread trash in elastic configurations with dynamically increasing and decreasing concurrent consumers. | | DefaultTaskExecutor Type | **jmsKeyFormatStrategy** (advanced) | 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. | | JmsKeyFormatStrategy +| **allowAdditionalHeaders** (producer) | This option is used to allow additional headers which may have values that are invalid according to JMS specification. For example some message systems such as WMQ do this vith headers JMS_IBM_MQMD_ that contains byte or other invalid types. You can specify multiple header names separated by comma and use as suffix for wildcard matching. | | String | **queueBrowseStrategy** (advanced) | To use a custom QueueBrowseStrategy when browsing queues | | QueueBrowseStrategy | **messageCreatedStrategy** (advanced) | To use the given MessageCreatedStrategy which are invoked when Camel creates new instances of javax.jms.Message objects when Camel is sending a JMS message. | | MessageCreatedStrategy | **waitForProvision CorrelationToBeUpdated Counter** (advanced) | Number of times to wait for provisional correlation id to be updated to the actual correlation id when doing request/reply over JMS and when the option useMessageIDAsCorrelationID is enabled. | 50 | int @@ -317,7 +318,7 @@ with the following path and query parameters: | **destinationName** | *Required* Name of the queue or topic to use as destination | | String |======================================================================= -#### Query Parameters (85 parameters): +#### Query Parameters (86 parameters): [width="100%",cols="2,5,^1,2",options="header"] |======================================================================= @@ -361,6 +362,7 @@ with the following path and query parameters: | **replyToType** (producer) | Allows for explicitly specifying which kind of strategy to use for replyTo queues when doing request/reply over JMS. Possible values are: Temporary Shared or Exclusive. By default Camel will use temporary queues. However if replyTo has been configured then Shared is used by default. This option allows you to use exclusive queues instead of shared ones. See Camel JMS documentation for more details and especially the notes about the implications if running in a clustered environment and the fact that Shared reply queues has lower performance than its alternatives Temporary and Exclusive. | | ReplyToType | **requestTimeout** (producer) | The timeout for waiting for a reply when using the InOut Exchange Pattern (in milliseconds). The default is 20 seconds. You can include the header CamelJmsRequestTimeout to override this endpoint configured timeout value and thus have per message individual timeout values. See also the requestTimeoutCheckerInterval option. | 20000 | long | **timeToLive** (producer) | When sending messages specifies the time-to-live of the message (in milliseconds). | -1 | long +| **allowAdditionalHeaders** (producer) | This option is used to allow additional headers which may have values that are invalid according to JMS specification. For example some message systems such as WMQ do this vith headers JMS_IBM_MQMD_ that contains byte or other invalid types. You can specify multiple header names separated by comma and use as suffix for wildcard matching. | | String | **allowNullBody** (producer) | Whether to allow sending messages with no body. If this option is false and the message body is null then an JMSException is thrown. | true | boolean | **alwaysCopyMessage** (producer) | If true Camel will always make a JMS message copy of the message when it is passed to the producer for sending. Copying the message is needed in some situations such as when a replyToDestinationSelectorName is set (incidentally Camel will set the alwaysCopyMessage option to true if a replyToDestinationSelectorName is set) | false | boolean | **correlationProperty** (producer) | Use this JMS property to correlate messages in InOut exchange pattern (request-reply) instead of JMSCorrelationID property. This allows you to exchange messages with systems that do not correlate messages using JMSCorrelationID JMS property. If used JMSCorrelationID will not be used or set by Camel. The value of here named property will be generated if not supplied in the header of the message under the same name. | | String http://git-wip-us.apache.org/repos/asf/camel/blob/39057c81/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java ---------------------------------------------------------------------- diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java index 871ef64..188921a 100644 --- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java +++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java @@ -26,6 +26,7 @@ import java.nio.ByteBuffer; import java.util.Date; import java.util.Enumeration; import java.util.HashMap; +import java.util.Iterator; import java.util.Map; import java.util.Map.Entry; import java.util.Set; @@ -40,6 +41,7 @@ import javax.jms.Session; import javax.jms.StreamMessage; import javax.jms.TextMessage; +import org.apache.camel.util.EndpointHelper; import org.w3c.dom.Node; import org.apache.camel.CamelContext; @@ -363,6 +365,18 @@ public class JmsBinding { // 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 the value was null, then it may be allowed as an additional header + if (value == null && endpoint.getConfiguration().getAllowAdditionalHeaders() != null) { + Iterator it = ObjectHelper.createIterator(endpoint.getConfiguration().getAllowAdditionalHeaders()); + while (it.hasNext()) { + String pattern = (String) it.next(); + if (EndpointHelper.matchPattern(headerName, pattern)) { + LOG.debug("Header {} allowed as additional header despite not being valid according to the JMS specification", headerName); + value = headerValue; + break; + } + } + } if (value != null) { // must encode to safe JMS header name before setting property on jmsMessage String key = jmsKeyFormatStrategy.encodeKey(headerName); @@ -414,6 +428,8 @@ public class JmsBinding { * <li>String and any other literals, Character, CharSequence</li> * <li>Boolean</li> * <li>Number</li> + * <li>java.math.BigInteger</li> + * <li>java.math.BigDecimal</li> * <li>java.util.Date</li> * </ul> * http://git-wip-us.apache.org/repos/asf/camel/blob/39057c81/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsComponent.java ---------------------------------------------------------------------- diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsComponent.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsComponent.java index cfe4ef5..8953020 100644 --- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsComponent.java +++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsComponent.java @@ -1050,6 +1050,19 @@ public class JmsComponent extends HeaderFilterStrategyComponent implements Appli } /** + * This option is used to allow additional headers which may have values that are invalid according to JMS specification. + + For example some message systems such as WMQ do this vith headers JMS_IBM_MQMD_* that contains byte[] or other invalid types. + + You can specify multiple header names separated by comma, and use * as suffix for wildcard matching. + */ + @Metadata(label = "producer,advanced", + description = "This option is used to allow additional headers which may have values that are invalid according to JMS specification." + + " For example some message systems such as WMQ do this vith headers JMS_IBM_MQMD_* that contains byte[] or other invalid types." + + " You can specify multiple header names separated by comma, and use * as suffix for wildcard matching.") + public void setAllowAdditionalHeaders(String allowAdditionalHeaders) { + getConfiguration().setAllowAdditionalHeaders(allowAdditionalHeaders); + } + + /** * Sets the Spring ApplicationContext to use */ public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { http://git-wip-us.apache.org/repos/asf/camel/blob/39057c81/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java ---------------------------------------------------------------------- diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java index bce4831..d72d645 100644 --- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java +++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java @@ -451,6 +451,11 @@ public class JmsConfiguration implements Cloneable { + " JMS property to correlate messages. If set messages will be correlated solely on the" + " value of this property JMSCorrelationID property will be ignored and not set by Camel.") private String correlationProperty; + @UriParam(label = "producer,advanced", + description = "This option is used to allow additional headers which may have values that are invalid according to JMS specification." + + " For example some message systems such as WMQ do this vith headers JMS_IBM_MQMD_* that contains byte[] or other invalid types." + + " You can specify multiple header names separated by comma, and use * as suffix for wildcard matching.") + private String allowAdditionalHeaders; public JmsConfiguration() { } @@ -2102,4 +2107,17 @@ public class JmsConfiguration implements Cloneable { public String getCorrelationProperty() { return correlationProperty; } + + public String getAllowAdditionalHeaders() { + return allowAdditionalHeaders; + } + + /** + * This option is used to allow additional headers which may have values that are invalid according to JMS specification. + + For example some message systems such as WMQ do this vith headers JMS_IBM_MQMD_* that contains byte[] or other invalid types. + + You can specify multiple header names separated by comma, and use * as suffix for wildcard matching. + */ + public void setAllowAdditionalHeaders(String allowAdditionalHeaders) { + this.allowAdditionalHeaders = allowAdditionalHeaders; + } } http://git-wip-us.apache.org/repos/asf/camel/blob/39057c81/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java ---------------------------------------------------------------------- diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java index a063a92..a1c631a 100644 --- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java +++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java @@ -1234,6 +1234,16 @@ public class JmsEndpoint extends DefaultEndpoint implements AsyncEndpoint, Heade configuration.setDefaultTaskExecutorType(type); } + @ManagedAttribute + public String getAllowAdditionalHeaders() { + return configuration.getAllowAdditionalHeaders(); + } + + @ManagedAttribute + public void setAllowAdditionalHeaders(String AllowAdditionalHeaders) { + configuration.setAllowAdditionalHeaders(AllowAdditionalHeaders); + } + public MessageListenerContainerFactory getMessageListenerContainerFactory() { return configuration.getMessageListenerContainerFactory(); } http://git-wip-us.apache.org/repos/asf/camel/blob/39057c81/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsAllowAdditionalHeadersTest.java ---------------------------------------------------------------------- diff --git a/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsAllowAdditionalHeadersTest.java b/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsAllowAdditionalHeadersTest.java new file mode 100644 index 0000000..1c62a04 --- /dev/null +++ b/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsAllowAdditionalHeadersTest.java @@ -0,0 +1,74 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.jms; + +import javax.jms.ConnectionFactory; + +import org.apache.camel.CamelContext; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.test.junit4.CamelTestSupport; +import org.junit.Test; + +import static org.apache.camel.component.jms.JmsComponent.jmsComponentAutoAcknowledge; + +/** + * @version + */ +public class JmsAllowAdditionalHeadersTest extends CamelTestSupport { + + @Test + public void testAllowAdditionalHeaders() throws Exception { + // byte[] data = "Camel Rocks".getBytes(); + Object data = "Camel Rocks"; + + getMockEndpoint("mock:bar").expectedBodiesReceived("Hello World"); + getMockEndpoint("mock:bar").expectedHeaderReceived("foo", "bar"); + // ActiveMQ will not accept byte[] value + // getMockEndpoint("mock:bar").expectedHeaderReceived("JMS_IBM_MQMD_USER", data); + + fluentTemplate.withBody("Hello World").withHeader("foo", "bar").withHeader("JMS_IBM_MQMD_USER", data) + .to("direct:start").send(); + + assertMockEndpointsSatisfied(); + } + + protected CamelContext createCamelContext() throws Exception { + CamelContext camelContext = super.createCamelContext(); + + ConnectionFactory connectionFactory = CamelJmsTestHelper.createConnectionFactory(); + + JmsComponent jms = jmsComponentAutoAcknowledge(connectionFactory); + // allow any of those special IBM headers (notice we use * as wildcard) + jms.setAllowAdditionalHeaders("JMS_IBM_MQMD*"); + + camelContext.addComponent("jms", jms); + + return camelContext; + } + + @Override + protected RouteBuilder createRouteBuilder() throws Exception { + return new RouteBuilder() { + @Override + public void configure() throws Exception { + from("direct:start").to("jms:queue:bar"); + + from("jms:queue:bar").to("mock:bar"); + } + }; + } +} http://git-wip-us.apache.org/repos/asf/camel/blob/39057c81/platforms/spring-boot/components-starter/camel-jms-starter/src/main/java/org/apache/camel/component/jms/springboot/JmsComponentConfiguration.java ---------------------------------------------------------------------- diff --git a/platforms/spring-boot/components-starter/camel-jms-starter/src/main/java/org/apache/camel/component/jms/springboot/JmsComponentConfiguration.java b/platforms/spring-boot/components-starter/camel-jms-starter/src/main/java/org/apache/camel/component/jms/springboot/JmsComponentConfiguration.java index 5252fd0..ef604c1 100644 --- a/platforms/spring-boot/components-starter/camel-jms-starter/src/main/java/org/apache/camel/component/jms/springboot/JmsComponentConfiguration.java +++ b/platforms/spring-boot/components-starter/camel-jms-starter/src/main/java/org/apache/camel/component/jms/springboot/JmsComponentConfiguration.java @@ -532,6 +532,14 @@ public class JmsComponentConfiguration @NestedConfigurationProperty private JmsKeyFormatStrategy jmsKeyFormatStrategy; /** + * This option is used to allow additional headers which may have values + * that are invalid according to JMS specification. For example some message + * systems such as WMQ do this vith headers JMS_IBM_MQMD_ that contains byte + * or other invalid types. You can specify multiple header names separated + * by comma and use as suffix for wildcard matching. + */ + private String allowAdditionalHeaders; + /** * To use a custom QueueBrowseStrategy when browsing queues */ @NestedConfigurationProperty @@ -1134,6 +1142,14 @@ public class JmsComponentConfiguration this.jmsKeyFormatStrategy = jmsKeyFormatStrategy; } + public String getAllowAdditionalHeaders() { + return allowAdditionalHeaders; + } + + public void setAllowAdditionalHeaders(String allowAdditionalHeaders) { + this.allowAdditionalHeaders = allowAdditionalHeaders; + } + public QueueBrowseStrategy getQueueBrowseStrategy() { return queueBrowseStrategy; } @@ -1809,6 +1825,15 @@ public class JmsComponentConfiguration * name. */ private String correlationProperty; + /** + * This option is used to allow additional headers which may have values + * that are invalid according to JMS specification. + For example some + * message systems such as WMQ do this vith headers JMS_IBM_MQMD_* that + * contains byte[] or other invalid types. + You can specify multiple + * header names separated by comma, and use * as suffix for wildcard + * matching. + */ + private String allowAdditionalHeaders; public ConsumerType getConsumerType() { return consumerType; @@ -2534,5 +2559,13 @@ public class JmsComponentConfiguration public void setCorrelationProperty(String correlationProperty) { this.correlationProperty = correlationProperty; } + + public String getAllowAdditionalHeaders() { + return allowAdditionalHeaders; + } + + public void setAllowAdditionalHeaders(String allowAdditionalHeaders) { + this.allowAdditionalHeaders = allowAdditionalHeaders; + } } } \ No newline at end of file
