org.apache.ignite.stream.jms11

Class JmsStreamer<T extends javax.jms.Message,K,V>

java.lang.Object

org.apache.ignite.stream.StreamAdapter<T,K,V>

org.apache.ignite.stream.jms11.JmsStreamer<T,K,V>

public class JmsStreamer<T extends javax.jms.Message,K,V>
extends StreamAdapter<T,K,V>

Streamer that consumes from a JMS destination and feeds key-value pairs into an IgniteDataStreamer instance.

This Streamer uses purely JMS semantics and it is not coupled with any JMS implementation. It uses MessageListener to receive messages. You must provide your broker's ConnectionFactory when creating a JmsStreamer.

You must also provide a MessageTransformer to convert the incoming message into cache entries.

This Streamer has many features:

• Consumes from queues or topics.
• For topics, it supports durable subscriptions.
• Concurrent consumers are supported via the threads parameter. When consuming from queues, this component will start as many Session objects with separate MessageListener instances each, therefore achieving native concurrency (in terms of the JMS standard).
  When consuming from topics, obviously we cannot start multiple threads as that would lead us to consume duplicate messages. Therefore, we achieve concurrency in a virtualized manner through an internal thread pool.
• Transacted sessions are supported through the transacted parameter.
• Batched consumption is possible via the batched parameter. Depending on the broker, this technique can provide a higher throughput as it decreases the amount of message acknowledgement round trips that are necessary, albeit at the expense possible duplicate messages (especially if an incident occurs in the middle of a transaction).
  Batches are committed when the batchClosureMillis time has elapsed, or when a Session has received at least batchClosureSize messages. Time-based closure fires with the specified frequency and applies to all Sessions in parallel. Size-based closure applies individually to each Session (as transactions are Session-bound in JMS, so it will fire when that Session has processed that many messages. Both options are compatible with each other, or you can disable either (see setter documentation), but not both.
• Can specify the destination with implementation-specific Destination objects or with names.

Field Summary

Fields inherited from class org.apache.ignite.stream.StreamAdapter

ignite

Constructor Summary

Constructors

Constructor and Description
JmsStreamer()

Method Summary

Modifier and Type	Method and Description
void	setBatchClosureMillis(long batchClosureMillis) When using batched consumption, sets the time in milliseconds that will elapse before a batch is committed.
void	setBatchClosureSize(int batchClosureSize) When using batched consumption, sets the amount of messages that will be received before a batch is committed.
void	setBatched(boolean batched) Batch consumption leverages JMS Transactions to minimise round trips to the broker.
void	setClientId(String clientId) Sets the client ID of the JMS Connection.
void	setConnectionFactory(javax.jms.ConnectionFactory connectionFactory) Sets the JMS ConnectionFactory.
void	setDestination(javax.jms.Destination destination) Sets the JMS Destination explicitly.
void	setDestinationName(String destinationName) Sets the name of the JMS destination to consume from.
void	setDestinationType(Class<? extends javax.jms.Destination> destinationType) Sets the type of the destination to create, when used in combination with setDestinationName(String).
void	setDurableSubscription(boolean durableSubscription) A true value is only accepted in combination with topic consumption.
void	setDurableSubscriptionName(String durableSubscriptionName) When using Durable Subscribers, sets the name of the durable subscriber.
void	setExceptionListener(javax.jms.ExceptionListener exceptionListener) Exception listener for queue/topic failures.
void	setThreads(int threads) Sets the number of threads to concurrently consume JMS messages.
void	setTransacted(boolean transacted) Instructs the streamer whether to use local JMS transactions or not.
void	setTransformer(MessageTransformer<T,K,V> transformer) Compulsory.
void	start() Starts streamer.
void	stop() Stops streamer.

Methods inherited from class org.apache.ignite.stream.StreamAdapter

addMessage, getIgnite, getMultipleTupleExtractor, getSingleTupleExtractor, getStreamer, getTupleExtractor, setIgnite, setMultipleTupleExtractor, setSingleTupleExtractor, setStreamer, setTupleExtractor

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

JmsStreamer

public JmsStreamer()

Method Detail

start

public void start()
           throws IgniteException

Starts streamer.

Throws:

IgniteException - If failed.

stop

public void stop()
          throws IgniteException

Stops streamer.

Throws:

IgniteException

setConnectionFactory

public void setConnectionFactory(javax.jms.ConnectionFactory connectionFactory)

Sets the JMS ConnectionFactory.

Parameters:

connectionFactory - JMS ConnectionFactory for this streamer to use.

setTransformer

public void setTransformer(MessageTransformer<T,K,V> transformer)

Compulsory. The MessageTransformer that converts an incoming JMS Message (or subclass) into one or multiple cache entries.

Parameters:

transformer - The implementation of the MessageTransformer to use.

setDestination

public void setDestination(javax.jms.Destination destination)

Sets the JMS Destination explicitly. Takes precedence over destinationName if both are set.

Parameters:

destination - JMS Destination if setting it explicitly.

setDestinationName

public void setDestinationName(String destinationName)

Sets the name of the JMS destination to consume from.

Parameters:

destinationName - The name of the destination; will be passed on directly to the broker.

setDestinationType

public void setDestinationType(Class<? extends javax.jms.Destination> destinationType)

Sets the type of the destination to create, when used in combination with setDestinationName(String). It can be an interface or the implementation class specific to the broker.

Parameters:

destinationType - The class representing the destination type. Suggested values: Queue or Topic. Compulsory if using destinationName.

See Also:

Queue, Topic

setThreads

public void setThreads(int threads)

Sets the number of threads to concurrently consume JMS messages.

When working with queues, we will start as many Session objects as indicated by this field, i.e. you will get native concurrency.

On the other hand, when consuming from a topic, for obvious reason we will only start 1 message consumer but we will distribute the processing of received messages to as many concurrent threads as indicated.

Parameters:

threads - Number of threads to use. Default: 1.

setClientId

public void setClientId(String clientId)

Sets the client ID of the JMS Connection.

Parameters:

clientId - Client ID in case we're using durable subscribers. Default: none.

setDurableSubscription

public void setDurableSubscription(boolean durableSubscription)

A true value is only accepted in combination with topic consumption.

Parameters:

durableSubscription - Whether or not to use durable subscriptions. Default: false.

setTransacted

public void setTransacted(boolean transacted)

Instructs the streamer whether to use local JMS transactions or not.

Parameters:

transacted - Whether to consume or not in a transacted manner. Default: false.

setBatched

public void setBatched(boolean batched)

Batch consumption leverages JMS Transactions to minimise round trips to the broker.

Rather than ACKing every single message received, they will be received in the context of a JMS transaction which will be committed once the indicated batch closure size or batch closure time has elapsed.

Warning: May lead to duplicate consumption.

Parameters:

batched - Whether to consume messages in batches. Value true implies transacted = true. Default: false.

See Also:

setBatchClosureMillis(long), setBatchClosureSize(int)

setBatchClosureSize

public void setBatchClosureSize(int batchClosureSize)

When using batched consumption, sets the amount of messages that will be received before a batch is committed.

Parameters:

batchClosureSize - The amount of messages processed before a batch is committed. Default: 50.

setBatchClosureMillis

public void setBatchClosureMillis(long batchClosureMillis)

When using batched consumption, sets the time in milliseconds that will elapse before a batch is committed.

Parameters:

batchClosureMillis - Milliseconds before a batch is committed. Default: 1000ms.

setDurableSubscriptionName

public void setDurableSubscriptionName(String durableSubscriptionName)

When using Durable Subscribers, sets the name of the durable subscriber. It is compulsory.

Parameters:

durableSubscriptionName - Name of the durable subscriber. Default: none.

setExceptionListener

public void setExceptionListener(javax.jms.ExceptionListener exceptionListener)

Exception listener for queue/topic failures.

Parameters:

exceptionListener - ExceptionListener interface implementation.