The JMS message model supports several types of message bodies in (see javax.jms.Message), which may be created directly from the Akka Stream elements, or in wrappers to access more advanced features.
JmsConsumerJmsConsumer offers factory methods to consume JMS messages in a number of ways.
This examples shows how to listen to a JMS queue and emit javax.jms.Message elements into the stream.
The materialized value JmsConsumerControl is used to shut down the consumer (it is a Killswitch) and offers the possibility to inspect the connectivity state of the consumer.
The JmsConsumerSettings factories allow for passing the actor system to read from the default alpakka.jms.consumer section, or you may pass a Config instance which is resolved to a section of the same structure.
The Alpakka JMS consumer is configured via default settings in the HOCON config file section alpakka.jms.consumer in your application.conf, and settings may be tweaked in the code using the withXyz methods. On the second tab the section from reference.conf shows the structure to use for configuring multiple set-ups.
# Number of parallel sessions to use for receiving JMS messages.
# (or acks are pending for acknowledged consumers).
To reach out to special features of the JMS broker, destinations can be created as CustomDestination which takes a factory method for creating destinations.
Client acknowledgement ensures a message is successfully received by the consumer and notifies the JMS broker for every message. Due to the threading details in JMS brokers, this special source is required (see the explanation below).
The sessionCount parameter controls the number of JMS sessions to run in parallel.
Using multiple sessions increases throughput, especially if acknowledging message by message is desired.
Messages may arrive out of order if sessionCount is larger than 1.
Message-by-message acknowledgement can be achieved by setting bufferSize to 0, thus disabling buffering. The outstanding messages before backpressure will be the sessionCount.
The default AcknowledgeMode is ClientAcknowledge but can be overridden to custom AcknowledgeModes, even implementation-specific ones by setting the AcknowledgeMode in the JmsConsumerSettings when creating the stream.
Using a regular JmsConsumer with AcknowledgeMode.ClientAcknowledge and using message.acknowledge() from the stream is not compliant with the JMS specification and can cause issues for some message brokers. message.acknowledge() in many cases acknowledges the session and not the message itself, contrary to what the API makes you believe.
Use this JmsConsumer.ackSource as shown above instead.
JMS transactions may be used with this connector. Be aware that transactions are a heavy-weight tool and may not perform very good.
The ackTimeout parameter controls the maximum time given to a message to be committed or rolled back. If the message times out it will automatically be rolled back. This is to prevent stream from starvation if the application fails to commit or rollback a message, or if the message errors out and the stream is resumed by a decider.
Higher throughput is achieved by increasing the sessionCount.
Messages will arrive out of order if sessionCount is larger than 1.
Buffering is not supported in transaction mode. The bufferSize is ignored.
The default AcknowledgeMode is SessionTransacted but can be overridden to custom AcknowledgeModes, even implementation-specific ones by setting the AcknowledgeMode in the JmsConsumerSettings when creating the stream.
The objectSource emits the received message body as deserialized JVM instance. As serialization may be a security concern, JMS clients require special configuration to allow this. The example shows how to configure ActiveMQ connection factory to support serialization. See ActiveMQ Security for more information on this.