Class JmsSpout
- java.lang.Object
-
- org.apache.storm.topology.base.BaseComponent
-
- org.apache.storm.topology.base.BaseRichSpout
-
- org.apache.storm.jms.spout.JmsSpout
-
- All Implemented Interfaces:
Serializable,ISpout,IComponent,IRichSpout
public class JmsSpout extends BaseRichSpout
A StormSpoutimplementation that listens to a JMS topic or queue and outputs tuples based on the messages it receives.JmsSpoutinstances rely onJmsProducerimplementations to obtain the JMSConnectionFactoryandDestinationobjects necessary to connect to a JMS topic/queue.When a
JmsSpoutreceives a JMS message, it delegates to an internalJmsTupleProducerinstance to create a Storm tuple from the incoming message.Typically, developers will supply a custom
JmsTupleProducerimplementation appropriate for the expected message content.- See Also:
- Serialized Form
-
-
Constructor Summary
Constructors Constructor Description JmsSpout()
-
Method Summary
All Methods Instance Methods Concrete Methods Modifier and Type Method Description voidack(Object msgId)Ack a successfully handled message by the matchingJmsMessageID.voidclose()Close thesessionandconnection.voiddeclareOutputFields(OutputFieldsDeclarer declarer)Use thetupleProducerto determine which fields are about to be emitted.voidfail(Object msgId)Fail an unsuccessfully handled message by itsJmsMessageID.Map<String,Object>getComponentConfiguration()Declare configuration specific to this component.intgetJmsAcknowledgeMode()Returns the JMS Session acknowledgement mode for the JMS session associated with this spout.protected javax.jms.SessiongetSession()Returns the currently active session.booleanisDistributed()Returns if the spout is distributed.voidnextTuple()Generate the next tuple from a message.voidopen(Map<String,Object> conf, TopologyContext context, SpoutOutputCollector spoutOutputCollector)ISpoutimplementation.voidsetDistributed(boolean isDistributed)Sets the "distributed" mode of this spout.voidsetIndividualAcks()Set if JMS vendor supports ack-ing individual messages.voidsetJmsAcknowledgeMode(int mode)Sets the JMS Session acknowledgement mode for the JMS session.voidsetJmsProvider(JmsProvider provider)SetjmsProvider.voidsetJmsTupleProducer(JmsTupleProducer producer)Set theJmsTupleProducerimplementation that will convertjavax.jms.Messageobject toorg.apache.storm.tuple.Valuesobjects to be emitted.-
Methods inherited from class org.apache.storm.topology.base.BaseRichSpout
activate, deactivate
-
-
-
-
Method Detail
-
setJmsAcknowledgeMode
public void setJmsAcknowledgeMode(int mode)
Sets the JMS Session acknowledgement mode for the JMS session.Possible values:
- javax.jms.Session.AUTO_ACKNOWLEDGE
- javax.jms.Session.CLIENT_ACKNOWLEDGE
- javax.jms.Session.DUPS_OK_ACKNOWLEDGE
Any other vendor specific modes are not supported.
- Parameters:
mode- JMS Session Acknowledgement mode
-
getJmsAcknowledgeMode
public int getJmsAcknowledgeMode()
Returns the JMS Session acknowledgement mode for the JMS session associated with this spout. Can be either of:Session.AUTO_ACKNOWLEDGESession.CLIENT_ACKNOWLEDGESession.DUPS_OK_ACKNOWLEDGESession.SESSION_TRANSACTED
- Returns:
- the int value of the acknowledgment mode.
-
setJmsProvider
public void setJmsProvider(JmsProvider provider)
SetjmsProvider.Set the
JmsProviderimplementation that this Spout will use to connect to a JMSjavax.jms.Desination- Parameters:
provider- the provider to use
-
setJmsTupleProducer
public void setJmsTupleProducer(JmsTupleProducer producer)
Set theJmsTupleProducerimplementation that will convertjavax.jms.Messageobject toorg.apache.storm.tuple.Valuesobjects to be emitted.- Parameters:
producer- the producer instance to use
-
setIndividualAcks
public void setIndividualAcks()
Set if JMS vendor supports ack-ing individual messages. The appropriate mode must be set via {setJmsAcknowledgeMode(int)}.
-
open
public void open(Map<String,Object> conf, TopologyContext context, SpoutOutputCollector spoutOutputCollector)
ISpoutimplementation.Connects the JMS spout to the configured JMS destination topic/queue.
- Parameters:
conf- The Storm configuration for this spout. This is the configuration provided to the topology merged in with cluster configuration on this machine.context- This object can be used to get information about this task's place within the topology, including the task id and component id of this task, input and output information, etc.spoutOutputCollector- The collector is used to emit tuples from this spout. Tuples can be emitted at any time, including the open and close methods. The collector is thread-safe and should be saved as an instance variable of this spout object.
-
close
public void close()
Close thesessionandconnection.When overridden, should always call
superto finalize the active connections.- Specified by:
closein interfaceISpout- Overrides:
closein classBaseRichSpout
-
nextTuple
public void nextTuple()
Generate the next tuple from a message.This method polls the queue that's being filled asynchronously by the jms connection, every
POLL_INTERVAL_MSseconds.
-
ack
public void ack(Object msgId)
Ack a successfully handled message by the matchingJmsMessageID.Acking means removing the message from the pending messages collections, and if it was the oldest pending message - ack it to the mq as well, so that it's the only one acked.
Will only be called if we're transactional or not AUTO_ACKNOWLEDGE.
- Specified by:
ackin interfaceISpout- Overrides:
ackin classBaseRichSpout
-
fail
public void fail(Object msgId)
Fail an unsuccessfully handled message by itsJmsMessageID.Failing means dropping all pending messages and queueing a recovery attempt.
Will only be called if we're transactional or not AUTO_ACKNOWLEDGE
- Specified by:
failin interfaceISpout- Overrides:
failin classBaseRichSpout
-
declareOutputFields
public void declareOutputFields(OutputFieldsDeclarer declarer)
Use thetupleProducerto determine which fields are about to be emitted.Note that
nextTuple()always emits to the default stream, and thus only fields declared for this stream are used.- Parameters:
declarer- this is used to declare output stream ids, output fields, and whether or not each output stream is a direct stream
-
isDistributed
public boolean isDistributed()
Returns if the spout is distributed.- Returns:
distributed.
-
setDistributed
public void setDistributed(boolean isDistributed)
Sets the "distributed" mode of this spout.If
truemultiple instances of this spout may be created across the cluster (depending on the "parallelism_hint" in the topology configuration).Setting this value to
falseessentially means this spout will run as a singleton within the cluster ("parallelism_hint" will be ignored).In general, this should be set to
falseif the underlying JMS destination is a topic, andtrueif it is a JMS queue.- Parameters:
isDistributed-trueif should be distributed,falseotherwise.
-
getSession
protected javax.jms.Session getSession()
Returns the currently active session.- Returns:
- The currently active session
-
getComponentConfiguration
public Map<String,Object> getComponentConfiguration()
Description copied from interface:IComponentDeclare configuration specific to this component. Only a subset of the "topology.*" configs can be overridden. The component configuration can be further overridden when constructing the topology usingTopologyBuilder- Specified by:
getComponentConfigurationin interfaceIComponent- Overrides:
getComponentConfigurationin classBaseComponent
-
-