| [[sjms2-component]] |
| = Simple JMS2 Component |
| |
| *Available as of Camel version 2.19* |
| |
| The Simple JMS 2.0 Component, or SJMS2, is a JMS client for use with Camel |
| that uses well known best practices when it comes to JMS client creation |
| and configuration. SJMS2 contains a brand new JMS 2.0 client API written |
| explicitly for Camel eliminating third party messaging implementations |
| keeping it light and resilient. The following features is included: |
| |
| * Standard Queue and Topic Support (Durable & Non-Durable) |
| * InOnly & InOut MEP Support |
| * Asynchronous Producer and Consumer Processing |
| * Internal JMS Transaction Support |
| |
| Additional key features include: |
| |
| * Plugable Connection Resource Management |
| * Session, Consumer, & Producer Pooling & Caching Management |
| * Batch Consumers and Producers |
| * Transacted Batch Consumers & Producers |
| * Support for Customizable Transaction Commit Strategies (Local JMS |
| Transactions only) |
| |
| [NOTE] |
| ==== |
| *Why the S in SJMS* |
| |
| S stands for Simple and Standard and Springless. Also camel-jms was |
| already taken. |
| ==== |
| |
| Maven users will need to add the following dependency to their `pom.xml` |
| for this component: |
| |
| [source,xml] |
| ---- |
| <dependency> |
| <groupId>org.apache.camel</groupId> |
| <artifactId>camel-sjms2</artifactId> |
| <version>x.x.x</version> |
| <!-- use the same version as your Camel core version --> |
| </dependency> |
| ---- |
| |
| == URI format |
| |
| [source] |
| ---- |
| sjms2:[queue:|topic:]destinationName[?options] |
| ---- |
| |
| Where `destinationName` is a JMS queue or topic name. By default, the |
| `destinationName` is interpreted as a queue name. For example, to |
| connect to the queue, `FOO.BAR` use: |
| |
| [source] |
| ---- |
| sjms2:FOO.BAR |
| ---- |
| |
| You can include the optional `queue:` prefix, if you prefer: |
| |
| [source] |
| ---- |
| sjms2:queue:FOO.BAR |
| ---- |
| |
| To connect to a topic, you _must_ include the `topic:` prefix. For |
| example, to connect to the topic, `Stocks.Prices`, use: |
| |
| [source] |
| ---- |
| sjms2:topic:Stocks.Prices |
| ---- |
| |
| You append query options to the URI using the following format, |
| `?option=value&option=value&...` |
| |
| == Component Options and Configurations |
| |
| |
| |
| |
| // component options: START |
| The Simple JMS2 component supports 15 options, which are listed below. |
| |
| |
| |
| [width="100%",cols="2,5,^1,2",options="header"] |
| |=== |
| | Name | Description | Default | Type |
| | *connectionFactory* (advanced) | A ConnectionFactory is required to enable the SjmsComponent. It can be set directly or set set as part of a ConnectionResource. | | ConnectionFactory |
| | *connectionResource* (advanced) | A ConnectionResource is an interface that allows for customization and container control of the ConnectionFactory. See Plugable Connection Resource Management for further details. | | ConnectionResource |
| | *connectionCount* (common) | The maximum number of connections available to endpoints started under this component | 1 | Integer |
| | *jmsKeyFormatStrategy* (advanced) | Pluggable strategy for encoding and decoding JMS keys so they can be compliant with the JMS specification. Camel provides one implementation out of the box: default. The default strategy will safely marshal dots and hyphens (. and -). 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 |
| | *transactionCommit Strategy* (transaction) | To configure which kind of commit strategy to use. Camel provides two implementations out of the box, default and batch. | | TransactionCommitStrategy |
| | *destinationCreation Strategy* (advanced) | To use a custom DestinationCreationStrategy. | | DestinationCreationStrategy |
| | *timedTaskManager* (advanced) | To use a custom TimedTaskManager | | TimedTaskManager |
| | *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 |
| | *connectionTestOnBorrow* (advanced) | When using the default org.apache.camel.component.sjms.jms.ConnectionFactoryResource then should each javax.jms.Connection be tested (calling start) before returned from the pool. | true | boolean |
| | *connectionUsername* (security) | The username to use when creating javax.jms.Connection when using the default org.apache.camel.component.sjms.jms.ConnectionFactoryResource. | | String |
| | *connectionPassword* (security) | The password to use when creating javax.jms.Connection when using the default org.apache.camel.component.sjms.jms.ConnectionFactoryResource. | | String |
| | *connectionClientId* (advanced) | The client ID to use when creating javax.jms.Connection when using the default org.apache.camel.component.sjms.jms.ConnectionFactoryResource. | | String |
| | *connectionMaxWait* (advanced) | The max wait time in millis to block and wait on free connection when the pool is exhausted when using the default org.apache.camel.component.sjms.jms.ConnectionFactoryResource. | 5000 | long |
| | *headerFilterStrategy* (filter) | To use a custom org.apache.camel.spi.HeaderFilterStrategy to filter header to and from Camel message. | | HeaderFilterStrategy |
| | *basicPropertyBinding* (advanced) | Whether the component should use basic property binding (Camel 2.x) or the newer property binding with additional capabilities | false | boolean |
| |=== |
| // component options: END |
| |
| |
| |
| |
| |
| |
| |
| |
| // endpoint options: START |
| The Simple JMS2 endpoint is configured using URI syntax: |
| |
| ---- |
| sjms2:destinationType:destinationName |
| ---- |
| |
| with the following path and query parameters: |
| |
| === Path Parameters (2 parameters): |
| |
| |
| [width="100%",cols="2,5,^1,2",options="header"] |
| |=== |
| | Name | Description | Default | Type |
| | *destinationType* | The kind of destination to use | queue | String |
| | *destinationName* | *Required* DestinationName is a JMS queue or topic name. By default, the destinationName is interpreted as a queue name. | | String |
| |=== |
| |
| |
| === Query Parameters (39 parameters): |
| |
| |
| [width="100%",cols="2,5,^1,2",options="header"] |
| |=== |
| | Name | Description | Default | Type |
| | *durable* (consumer) | Sets topic consumer to durable. | false | boolean |
| | *shared* (consumer) | Sets the consumer to shared. | false | boolean |
| | *subscriptionId* (consumer) | Sets the subscription Id, required for durable or shared topics. | | String |
| | *acknowledgementMode* (common) | The JMS acknowledgement name, which is one of: SESSION_TRANSACTED, CLIENT_ACKNOWLEDGE, AUTO_ACKNOWLEDGE, DUPS_OK_ACKNOWLEDGE | AUTO_ACKNOWLEDGE | SessionAcknowledgementType |
| | *bridgeErrorHandler* (consumer) | Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored. | false | boolean |
| | *consumerCount* (consumer) | Sets the number of consumer listeners used for this endpoint. | 1 | int |
| | *durableSubscriptionId* (consumer) | Sets the durable subscription Id required for durable topics. | | String |
| | *synchronous* (consumer) | Sets whether synchronous processing should be strictly used or Camel is allowed to use asynchronous processing (if supported). | true | boolean |
| | *exceptionHandler* (consumer) | To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By default the consumer will deal with exceptions, that will be logged at WARN or ERROR level and ignored. | | ExceptionHandler |
| | *exchangePattern* (consumer) | Sets the exchange pattern when the consumer creates an exchange. | | ExchangePattern |
| | *messageSelector* (consumer) | Sets the JMS Message selector syntax. | | String |
| | *lazyStartProducer* (producer) | Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel's routing error handlers. Beware that when the first message is processed then creating and starting the producer may take a little time and prolong the total processing time of the processing. | false | boolean |
| | *namedReplyTo* (producer) | Sets the reply to destination name used for InOut producer endpoints. The type of the reply to destination can be determined by the starting prefix (topic: or queue:) in its name. | | String |
| | *persistent* (producer) | Flag used to enable/disable message persistence. | true | boolean |
| | *producerCount* (producer) | Sets the number of producers used for this endpoint. | 1 | int |
| | *ttl* (producer) | Flag used to adjust the Time To Live value of produced messages. | -1 | long |
| | *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 |
| | *prefillPool* (producer) | Whether to prefill the producer connection pool on startup, or create connections lazy when needed. | true | boolean |
| | *responseTimeOut* (producer) | Sets the amount of time we should wait before timing out a InOut response. | 5000 | long |
| | *asyncStartListener* (advanced) | Whether to startup the consumer message listener asynchronously, when starting a route. For example if a JmsConsumer cannot get a connection to a remote JMS broker, then it may block while retrying and/or failover. This will cause Camel to block while starting routes. By setting this option to true, you will let routes startup, while the JmsConsumer connects to the JMS broker using a dedicated thread in asynchronous mode. If this option is used, then beware that if the connection could not be established, then an exception is logged at WARN level, and the consumer will not be able to receive messages; You can then restart the route to retry. | false | boolean |
| | *asyncStopListener* (advanced) | Whether to stop the consumer message listener asynchronously, when stopping a route. | false | boolean |
| | *basicPropertyBinding* (advanced) | Whether the endpoint should use basic property binding (Camel 2.x) or the newer property binding with additional capabilities | false | boolean |
| | *connectionCount* (advanced) | The maximum number of connections available to this endpoint | | Integer |
| | *connectionFactory* (advanced) | Initializes the connectionFactory for the endpoint, which takes precedence over the component's connectionFactory, if any | | ConnectionFactory |
| | *connectionResource* (advanced) | Initializes the connectionResource for the endpoint, which takes precedence over the component's connectionResource, if any | | ConnectionResource |
| | *destinationCreationStrategy* (advanced) | To use a custom DestinationCreationStrategy. | | DestinationCreationStrategy |
| | *exceptionListener* (advanced) | Specifies the JMS Exception Listener that is to be notified of any underlying JMS exceptions. | | ExceptionListener |
| | *headerFilterStrategy* (advanced) | To use a custom HeaderFilterStrategy to filter header to and from Camel message. | | HeaderFilterStrategy |
| | *includeAllJMSXProperties* (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 |
| | *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 |
| | *mapJmsMessage* (advanced) | 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. | true | boolean |
| | *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 |
| | *errorHandlerLoggingLevel* (logging) | Allows to configure the default errorHandler logging level for logging uncaught exceptions. | WARN | LoggingLevel |
| | *errorHandlerLogStackTrace* (logging) | Allows to control whether stacktraces should be logged or not, by the default errorHandler. | true | boolean |
| | *transacted* (transaction) | Specifies whether to use transacted mode | false | boolean |
| | *transactionBatchCount* (transaction) | If transacted sets the number of messages to process before committing a transaction. | -1 | int |
| | *transactionBatchTimeout* (transaction) | Sets timeout (in millis) for batch transactions, the value should be 1000 or higher. | 5000 | long |
| | *transactionCommitStrategy* (transaction) | Sets the commit strategy. | | TransactionCommitStrategy |
| | *sharedJMSSession* (transaction) | Specifies whether to share JMS session with other SJMS endpoints. Turn this off if your route is accessing to multiple JMS providers. If you need transaction against multiple JMS providers, use jms component to leverage XA transaction. | true | boolean |
| |=== |
| // endpoint options: END |
| // spring-boot-auto-configure options: START |
| == Spring Boot Auto-Configuration |
| |
| When using Spring Boot make sure to use the following Maven dependency to have support for auto configuration: |
| |
| [source,xml] |
| ---- |
| <dependency> |
| <groupId>org.apache.camel</groupId> |
| <artifactId>camel-sjms2-starter</artifactId> |
| <version>x.x.x</version> |
| <!-- use the same version as your Camel core version --> |
| </dependency> |
| ---- |
| |
| |
| The component supports 16 options, which are listed below. |
| |
| |
| |
| [width="100%",cols="2,5,^1,2",options="header"] |
| |=== |
| | Name | Description | Default | Type |
| | *camel.component.sjms2.basic-property-binding* | Whether the component should use basic property binding (Camel 2.x) or the newer property binding with additional capabilities | false | Boolean |
| | *camel.component.sjms2.connection-client-id* | The client ID to use when creating javax.jms.Connection when using the default org.apache.camel.component.sjms.jms.ConnectionFactoryResource. | | String |
| | *camel.component.sjms2.connection-count* | The maximum number of connections available to endpoints started under this component | 1 | Integer |
| | *camel.component.sjms2.connection-factory* | A ConnectionFactory is required to enable the SjmsComponent. It can be set directly or set set as part of a ConnectionResource. The option is a javax.jms.ConnectionFactory type. | | String |
| | *camel.component.sjms2.connection-max-wait* | The max wait time in millis to block and wait on free connection when the pool is exhausted when using the default org.apache.camel.component.sjms.jms.ConnectionFactoryResource. | 5000 | Long |
| | *camel.component.sjms2.connection-password* | The password to use when creating javax.jms.Connection when using the default org.apache.camel.component.sjms.jms.ConnectionFactoryResource. | | String |
| | *camel.component.sjms2.connection-resource* | A ConnectionResource is an interface that allows for customization and container control of the ConnectionFactory. See Plugable Connection Resource Management for further details. The option is a org.apache.camel.component.sjms.jms.ConnectionResource type. | | String |
| | *camel.component.sjms2.connection-test-on-borrow* | When using the default org.apache.camel.component.sjms.jms.ConnectionFactoryResource then should each javax.jms.Connection be tested (calling start) before returned from the pool. | true | Boolean |
| | *camel.component.sjms2.connection-username* | The username to use when creating javax.jms.Connection when using the default org.apache.camel.component.sjms.jms.ConnectionFactoryResource. | | String |
| | *camel.component.sjms2.destination-creation-strategy* | To use a custom DestinationCreationStrategy. The option is a org.apache.camel.component.sjms.jms.DestinationCreationStrategy type. | | String |
| | *camel.component.sjms2.enabled* | Enable sjms2 component | true | Boolean |
| | *camel.component.sjms2.header-filter-strategy* | To use a custom org.apache.camel.spi.HeaderFilterStrategy to filter header to and from Camel message. The option is a org.apache.camel.spi.HeaderFilterStrategy type. | | String |
| | *camel.component.sjms2.jms-key-format-strategy* | Pluggable strategy for encoding and decoding JMS keys so they can be compliant with the JMS specification. Camel provides one implementation out of the box: default. The default strategy will safely marshal dots and hyphens (. and -). 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. The option is a org.apache.camel.component.sjms.jms.JmsKeyFormatStrategy type. | | String |
| | *camel.component.sjms2.message-created-strategy* | 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. The option is a org.apache.camel.component.sjms.jms.MessageCreatedStrategy type. | | String |
| | *camel.component.sjms2.timed-task-manager* | To use a custom TimedTaskManager. The option is a org.apache.camel.component.sjms.taskmanager.TimedTaskManager type. | | String |
| | *camel.component.sjms2.transaction-commit-strategy* | To configure which kind of commit strategy to use. Camel provides two implementations out of the box, default and batch. The option is a org.apache.camel.component.sjms.TransactionCommitStrategy type. | | String |
| |=== |
| // spring-boot-auto-configure options: END |
| |
| |
| |
| |
| |
| |
| Below is an example of how to configure the `Sjms2Component` with its |
| required `ConnectionFactory` provider. It will create a single connection |
| by default and store it using the component's internal pooling APIs to |
| ensure that it is able to service Session creation requests in a thread |
| safe manner. |
| |
| [source,java] |
| ---- |
| Sjms2Component component = new Sjms2Component(); |
| component.setConnectionFactory(new ActiveMQConnectionFactory("tcp://localhost:61616")); |
| getContext().addComponent("sjms2", component); |
| ---- |
| |
| For a SJMS2 component that is required to support a durable subscription, |
| you can override the default `ConnectionFactoryResource` instance and set |
| the `clientId` property. |
| |
| [source,java] |
| ---- |
| ConnectionFactoryResource connectionResource = new ConnectionFactoryResource(); |
| connectionResource.setConnectionFactory(new ActiveMQConnectionFactory("tcp://localhost:61616")); |
| connectionResource.setClientId("myclient-id"); |
| |
| Sjms2Component component = new Sjms2Component(); |
| component.setConnectionResource(connectionResource); |
| component.setMaxConnections(1); |
| ---- |
| |
| == Producer Usage |
| |
| === InOnly Producer - (Default) |
| |
| The _InOnly_ producer is the default behavior of the SJMS2 Producer |
| Endpoint. |
| |
| [source,java] |
| ---- |
| from("direct:start") |
| .to("sjms2:queue:bar"); |
| ---- |
| |
| === InOut Producer |
| |
| To enable _InOut_ behavior append the `exchangePattern` attribute to the |
| URI. By default it will use a dedicated TemporaryQueue for each |
| consumer. |
| |
| [source,java] |
| ---- |
| from("direct:start") |
| .to("sjms2:queue:bar?exchangePattern=InOut"); |
| ---- |
| |
| You can specify a `namedReplyTo` though which can provide a better |
| monitor point. |
| |
| [source,java] |
| ---- |
| from("direct:start") |
| .to("sjms2:queue:bar?exchangePattern=InOut&namedReplyTo=my.reply.to.queue"); |
| ---- |
| |
| == Consumer Usage |
| |
| === Durable Shared Subscription |
| |
| To create a durable subscription that can be shared between one or more consumers. |
| Use a JMS 2.0 compliant connection factory and specify a common subscriptionId. Then set the subscription properties durable and shared to true. |
| |
| [source,java] |
| ---- |
| from("sjms2:topic:foo?consumerCount=3&subscriptionId=bar&durable=true&shared=true") |
| .to("mock:result"); |
| |
| from("sjms2:topic:foo?consumerCount=2&subscriptionId=bar&durable=true&shared=true") |
| .to("mock:result"); |
| ---- |
| |
| === InOnly Consumer - (Default) |
| |
| The _InOnly_ xonsumer is the default Exchange behavior of the SJMS2 |
| Consumer Endpoint. |
| |
| [source,java] |
| ---- |
| from("sjms2:queue:bar") |
| .to("mock:result"); |
| ---- |
| |
| === InOut Consumer |
| |
| To enable _InOut_ behavior append the `exchangePattern` attribute to the |
| URI. |
| |
| [source,java] |
| ---- |
| from("sjms2:queue:in.out.test?exchangePattern=InOut") |
| .transform(constant("Bye Camel")); |
| ---- |
| |
| == Advanced Usage Notes |
| |
| === Plugable Connection Resource Management [[SJMS2-connectionresource]] |
| |
| SJMS2 provides JMS |
| http://docs.oracle.com/javaee/5/api/javax/jms/Connection.html[`Connection`] |
| resource management through built-in connection pooling. This eliminates |
| the need to depend on third party API pooling logic. However there may |
| be times that you are required to use an external Connection resource |
| manager such as those provided by J2EE or OSGi containers. For this SJMS2 |
| provides an interface that can be used to override the internal SJMS2 |
| Connection pooling capabilities. This is accomplished through the |
| https://svn.apache.org/repos/asf/camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/ConnectionResource.java[`ConnectionResource`] |
| interface. |
| |
| The |
| https://svn.apache.org/repos/asf/camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/ConnectionResource.java[`ConnectionResource`] |
| provides methods for borrowing and returning Connections as needed is |
| the contract used to provide |
| http://docs.oracle.com/javaee/5/api/javax/jms/Connection.html[`Connection`] |
| pools to the SJMS2 component. A user should use when it is necessary to |
| integrate SJMS2 with an external connection pooling manager. |
| |
| It is recommended though that for standard |
| http://docs.oracle.com/javaee/5/api/javax/jms/ConnectionFactory.html[`ConnectionFactory`] |
| providers you use the |
| https://svn.apache.org/repos/asf/camel/trunk/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/it/ConnectionResourceIT.java[`ConnectionFactoryResource`] |
| implementation that is provided with SJMS2 as-is or extend as it is |
| optimized for this component. |
| |
| Below is an example of using the plugable ConnectionResource with the |
| ActiveMQ `PooledConnectionFactory`: |
| |
| [source,java] |
| ---- |
| public class AMQConnectionResource implements ConnectionResource { |
| private PooledConnectionFactory pcf; |
| |
| public AMQConnectionResource(String connectString, int maxConnections) { |
| super(); |
| pcf = new PooledConnectionFactory(connectString); |
| pcf.setMaxConnections(maxConnections); |
| pcf.start(); |
| } |
| |
| public void stop() { |
| pcf.stop(); |
| } |
| |
| @Override |
| public Connection borrowConnection() throws Exception { |
| Connection answer = pcf.createConnection(); |
| answer.start(); |
| return answer; |
| } |
| |
| @Override |
| public Connection borrowConnection(long timeout) throws Exception { |
| // SNIPPED... |
| } |
| |
| @Override |
| public void returnConnection(Connection connection) throws Exception { |
| // Do nothing since there isn't a way to return a Connection |
| // to the instance of PooledConnectionFactory |
| log.info("Connection returned"); |
| } |
| } |
| ---- |
| |
| Then pass in the `ConnectionResource` to the `Sjms2Component`: |
| |
| [source,java] |
| ---- |
| CamelContext camelContext = new DefaultCamelContext(); |
| AMQConnectionResource pool = new AMQConnectionResource("tcp://localhost:33333", 1); |
| Sjms2Component component = new Sjms2Component(); |
| component.setConnectionResource(pool); |
| camelContext.addComponent("sjms2", component); |
| ---- |
| |
| To see the full example of its usage please refer to the |
| https://svn.apache.org/repos/asf/camel/trunk/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/it/ConnectionResourceIT.java[`ConnectionResourceIT`]. |
| |
| === Session, Consumer, & Producer Pooling & Caching Management |
| |
| Coming soon ... |
| |
| === Batch Message Support |
| |
| The Sjms2Producer supports publishing a collection of messages by |
| creating an Exchange that encapsulates a `List`. This Sjms2Producer will |
| take then iterate through the contents of the List and publish each |
| message individually. |
| |
| If when producing a batch of messages there is the need to set headers |
| that are unique to each message you can use the SJMS2 |
| https://svn.apache.org/repos/asf/camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/BatchMessage.java[`BatchMessage`] |
| class. When the Sjms2Producer encounters a `BatchMessage` list it will |
| iterate each `BatchMessage` and publish the included payload and headers. |
| |
| Below is an example of using the BatchMessage class. First we create a |
| list of `BatchMessage`: |
| |
| [source,java] |
| ---- |
| List<BatchMessage<String>> messages = new ArrayList<BatchMessage<String>>(); |
| for (int i = 1; i <= messageCount; i++) { |
| String body = "Hello World " + i; |
| BatchMessage<String> message = new BatchMessage<String>(body, null); |
| messages.add(message); |
| } |
| ---- |
| |
| Then publish the list: |
| |
| [source,java] |
| ---- |
| template.sendBody("sjms2:queue:batch.queue", messages); |
| ---- |
| |
| === Customizable Transaction Commit Strategies (Local JMS Transactions only) |
| |
| SJMS2 provides a developer the means to create a custom and plugable |
| transaction strategy through the use of the |
| https://svn.apache.org/repos/asf/camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/TransactionCommitStrategy.java[`TransactionCommitStrategy`] |
| interface. This allows a user to define a unique set of circumstances |
| that the |
| https://svn.apache.org/repos/asf/camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/tx/SessionTransactionSynchronization.java[`SessionTransactionSynchronization`] |
| will use to determine when to commit the Session. An example of its use |
| is the |
| https://svn.apache.org/repos/asf/camel/trunk/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/tx/BatchTransactionCommitStrategy.java[`BatchTransactionCommitStrategy`] |
| which is detailed further in the next section. |
| |
| === Transacted Batch Consumers & Producers |
| |
| The SJMS2 component has been designed to support the batching of local JMS |
| transactions on both the Producer and Consumer endpoints. How they are |
| handled on each is very different though. |
| |
| The SJMS2 consumer endpoint is a straightforward implementation that will |
| process X messages before committing them with the associated Session. |
| To enable batched transaction on the consumer first enable transactions |
| by setting the `transacted` parameter to true and then adding the |
| `transactionBatchCount` and setting it to any value that is greater than |
| 0. For example the following configuration will commit the Session every |
| 10 messages: |
| |
| [source] |
| ---- |
| sjms2:queue:transacted.batch.consumer?transacted=true&transactionBatchCount=10 |
| ---- |
| |
| If an exception occurs during the processing of a batch on the consumer |
| endpoint, the Session rollback is invoked causing the messages to be |
| redelivered to the next available consumer. The counter is also reset to |
| 0 for the `BatchTransactionCommitStrategy` for the associated Session as |
| well. It is the responsibility of the user to ensure they put hooks in |
| their processors of batch messages to watch for messages with the |
| JMSRedelivered header set to true. This is the indicator that messages |
| were rolled back at some point and that a verification of a successful |
| processing should occur. |
| |
| A transacted batch consumer also carries with it an instance of an |
| internal timer that waits a default amount of time (5000ms) between |
| messages before committing the open transactions on the Session. The |
| default value of 5000ms (minimum of 1000ms) should be adequate for most |
| use-cases but if further tuning is necessary simply set the |
| `transactionBatchTimeout` parameter. |
| |
| [source] |
| ---- |
| sjms2:queue:transacted.batch.consumer?transacted=true&transactionBatchCount=10&transactionBatchTimeout=2000 |
| ---- |
| |
| The minimal value that will be accepted is 1000ms as the amount of |
| context switching may cause unnecessary performance impacts without |
| gaining benefit. |
| |
| The producer endpoint is handled much differently though. With the |
| producer after each message is delivered to its destination the Exchange |
| is closed and there is no longer a reference to that message. To make a |
| available all the messages available for redelivery you simply enable |
| transactions on a Producer Endpoint that is publishing BatchMessages. |
| The transaction will commit at the conclusion of the exchange which |
| includes all messages in the batch list. Nothing additional need be |
| configured. For example: |
| |
| [source,java] |
| ---- |
| List<BatchMessage<String>> messages = new ArrayList<BatchMessage<String>>(); |
| for (int i = 1; i <= messageCount; i++) { |
| String body = "Hello World " + i; |
| BatchMessage<String> message = new BatchMessage<String>(body, null); |
| messages.add(message); |
| } |
| ---- |
| |
| Now publish the List with transactions enabled: |
| |
| [source,java] |
| ---- |
| template.sendBody("sjms2:queue:batch.queue?transacted=true", messages); |
| ---- |
| |
| == Additional Notes |
| |
| === Message Header Format |
| |
| The SJMS2 Component uses the same header format strategy that is used in |
| the Camel JMS Component. This plugable strategy ensures that messages |
| sent over the wire conform to the JMS Message spec. |
| |
| For the `exchange.in.header` the following rules apply for the header |
| keys: |
| |
| * Keys starting with `JMS` or `JMSX` are reserved. |
| * `exchange.in.headers` keys must be literals and all be valid Java |
| identifiers (do not use dots in the key name). |
| * Camel replaces dots & hyphens and the reverse when when consuming JMS |
| messages: |
| |
| ** is replaced by _DOT_ and the reverse replacement when Camel consumes |
| the message. |
| ** is replaced by _HYPHEN_ and the reverse replacement when Camel |
| consumes the message. + |
| See also the option `jmsKeyFormatStrategy`, which allows use of your own |
| custom strategy for formatting keys. |
| |
| For the `exchange.in.header`, the following rules apply for the header |
| values: |
| |
| === Message Content |
| |
| To deliver content over the wire we must ensure that the body of the |
| message that is being delivered adheres to the JMS Message |
| Specification. Therefore, all that are produced must either be |
| primitives or their counter objects (such as `Integer`, `Long`, `Character`). |
| The types, `String`, `CharSequence`, `Date`, `BigDecimal` and `BigInteger` are all |
| converted to their `toString()` representation. All other types are |
| dropped. |
| |
| === Clustering |
| |
| When using _InOut_ with SJMS2 in a clustered environment you must either |
| use TemporaryQueue destinations or use a unique named reply to |
| destination per InOut producer endpoint. Message correlation is handled |
| by the endpoint, not with message selectors at the broker. The InOut |
| Producer Endpoint uses Java Concurrency Exchangers cached by the Message |
| `JMSCorrelationID`. This provides a nice performance increase while |
| reducing the overhead on the broker since all the messages are consumed |
| from the destination in the order they are produced by the interested |
| consumer. |
| |
| Currently the only correlation strategy is to use the `JMSCorrelationId`. |
| The _InOut_ Consumer uses this strategy as well ensuring that all |
| responses messages to the included `JMSReplyTo` destination also have the |
| `JMSCorrelationId` copied from the request as well. |
| |
| == Transaction Support [[SJMS2-transactions]] |
| |
| SJMS2 currently only supports the use of internal JMS Transactions. There |
| is no support for the Camel Transaction Processor or the Java |
| Transaction API (JTA). |
| |
| === Does Springless Mean I Can't Use Spring? |
| |
| Not at all. Below is an example of the SJMS2 component using the Spring |
| DSL: |
| |
| [source,java] |
| ---- |
| <route |
| id="inout.named.reply.to.producer.route"> |
| <from |
| uri="direct:invoke.named.reply.to.queue" /> |
| <to |
| uri="sjms2:queue:named.reply.to.queue?namedReplyTo=my.response.queue&exchangePattern=InOut" /> |
| </route> |
| ---- |
| |
| Springless refers to moving away from the dependency on the Spring JMS |
| API. A new JMS client API is being developed from the ground up to power |
| SJMS2. |
