Class JmsSpout
- All Implemented Interfaces:
Serializable,ISpout,IComponent,IRichSpout
Spout implementation that listens to a JMS topic or
queue and outputs tuples based on the messages it receives.
JmsSpout instances rely on JmsProducer
implementations to obtain the JMS
ConnectionFactory and Destination objects necessary
to connect to a JMS topic/queue.
When a JmsSpout receives a JMS message, it delegates to an
internal JmsTupleProducer instance to create a Storm tuple from
the incoming message.
Typically, developers will supply a custom JmsTupleProducer
implementation appropriate for the expected message content.
- See Also:
-
Constructor Summary
Constructors -
Method Summary
Modifier and TypeMethodDescriptionvoidAck a successfully handled message by the matchingJmsMessageID.voidclose()Close thesessionandconnection.voiddeclareOutputFields(OutputFieldsDeclarer declarer) Use thetupleProducerto determine which fields are about to be emitted.voidFail an unsuccessfully handled message by itsJmsMessageID.Declare configuration specific to this component.intReturns the JMS Session acknowledgement mode for the JMS session associated with this spout.protected javax.jms.SessionReturns the currently active session.booleanReturns if the spout is distributed.voidGenerate 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.voidSet 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
-
Constructor Details
-
JmsSpout
public JmsSpout()
-
-
Method Details
-
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
SetjmsProvider.Set the
JmsProviderimplementation that this Spout will use to connect to a JMSjavax.jms.Desination- Parameters:
provider- the provider to use
-
setJmsTupleProducer
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
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
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
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
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
-