Github user arunmahadevan commented on a diff in the pull request:
https://github.com/apache/storm/pull/2639#discussion_r187496062
--- Diff:
external/storm-jms/src/main/java/org/apache/storm/jms/spout/JmsSpout.java ---
@@ -18,164 +18,124 @@
package org.apache.storm.jms.spout;
-import java.io.Serializable;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Timer;
-import java.util.TimerTask;
-import java.util.TreeSet;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
-import javax.jms.Connection;
-import javax.jms.ConnectionFactory;
-import javax.jms.Destination;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageListener;
-import javax.jms.Session;
-import org.apache.storm.Config;
import org.apache.storm.jms.JmsProvider;
import org.apache.storm.jms.JmsTupleProducer;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Values;
-import org.apache.storm.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.Session;
+import java.util.HashMap;
+import java.util.Map;
+
/**
- * A Storm <code>Spout</code> implementation that listens to a JMS topic
or queue and outputs tuples based on the messages it receives.
+ * A Storm <code>Spout</code> implementation that listens to a JMS topic or
+ * queue and outputs tuples based on the messages it receives.
*
* <p><code>JmsSpout</code> instances rely on <code>JmsProducer</code>
* implementations to obtain the JMS
* <code>ConnectionFactory</code> and <code>Destination</code> objects
necessary
* to connect to a JMS topic/queue.
*
- * <p>When a <code>JmsSpout</code> receives a JMS message, it delegates to
an
- * internal <code>JmsTupleProducer</code> instance to create a Storm tuple
from the incoming message.
+ * <p>When a {@code JmsSpout} receives a JMS message, it delegates to an
+ * internal {@code JmsTupleProducer} instance to create a Storm tuple from
+ * the incoming message.
*
* <p>Typically, developers will supply a custom
<code>JmsTupleProducer</code>
* implementation appropriate for the expected message content.
*/
@SuppressWarnings("serial")
-public class JmsSpout extends BaseRichSpout implements MessageListener {
+public class JmsSpout extends BaseRichSpout {
- /**
- * The logger object instance for this class.
- */
+ /** The logger object instance for this class. */
private static final Logger LOG =
LoggerFactory.getLogger(JmsSpout.class);
- /**
- * The logger of the recovery task.
- */
- private static final Logger RECOVERY_TASK_LOG =
LoggerFactory.getLogger(RecoveryTask.class);
-
- /**
- * Time to sleep between queue polling attempts.
- */
+ /** Time to sleep between queue polling attempts. */
private static final int POLL_INTERVAL_MS = 50;
- /**
- * The default value for {@link Config#TOPOLOGY_MESSAGE_TIMEOUT_SECS}.
- */
- private static final int DEFAULT_MESSAGE_TIMEOUT_SECS = 30;
-
- /**
- * Time to wait before queuing the first recovery task.
- */
- private static final int RECOVERY_DELAY_MS = 10;
- /**
- * Used to safely recover failed JMS sessions across instances.
- */
- private final Serializable recoveryMutex = "RECOVERY_MUTEX";
/**
* The acknowledgment mode used for this instance.
*
* @see Session
*/
private int jmsAcknowledgeMode = Session.AUTO_ACKNOWLEDGE;
- /**
- * Indicates whether or not this spout should run as a singleton.
- */
+
+ /** Sets up the way we want to handle the emit, ack and fails. */
+ private transient MessageHandler messageHandler = new MessageHandler();
--- End diff --
Good catch. Yes I should make the MessageHandler serializable.
---
