Use a factory to create VirtualSpout instances

src/main/java/com/salesforce/storm/spout/dynamic/DelegateSpoutFactory.java

@@ -0,0 +1,71 @@
+/**
+ * Copyright (c) 2017, Salesforce.com, Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without modification, are permitted provided that the
+ * following conditions are met:
+ *
+ * * Redistributions of source code must retain the above copyright notice, this list of conditions and the following
+ *   disclaimer.
+ *
+ * * Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following
+ *   disclaimer in the documentation and/or other materials provided with the distribution.
+ *
+ * * Neither the name of Salesforce.com nor the names of its contributors may be used to endorse or promote products
+ *   derived from this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
+ * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+ * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+ * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
+ * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
+ * USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package com.salesforce.storm.spout.dynamic;
+
+import com.salesforce.storm.spout.dynamic.consumer.ConsumerState;
+import com.salesforce.storm.spout.dynamic.metrics.MetricsRecorder;
+
+/**
+ * Factory for easily creating {@link DelegateSpout} instances.
+ *
+ * Handy in things like {@link com.salesforce.storm.spout.dynamic.handler.SpoutHandler} where we want to abstract away particulars, such
+ * as the things passed down from {@link DynamicSpout} such as the {@link MetricsRecorder} as an example.
+ */
+public interface DelegateSpoutFactory {
+
+    /**
+     * Create a {@link DelegateSpout} instance.
+     * @param identifier identifier to use for this instance.
+     * @return instance of {@link DelegateSpout}.
+     */
+    <T extends DelegateSpout> T create(
+        final VirtualSpoutIdentifier identifier
+    );
+
+    /**
+     * Create a {@link DelegateSpout} instance.
+     * @param identifier identifier to use for this instance.
+     * @param startingState starting consumer state for this instance.
+     * @return instance of {@link DelegateSpout}.
+     */
+    <T extends DelegateSpout> T create(
+        final VirtualSpoutIdentifier identifier,
+        final ConsumerState startingState
+    );
+
+    /**
+     * Create a {@link DelegateSpout} instance.
+     * @param identifier identifier to use for this instance.
+     * @param startingState starting consumer state for this instance.
+     * @param endingState ending consumer state for this instance.
+     * @return instance of {@link DelegateSpout}.
+     */
+    <T extends DelegateSpout> T create(
+        final VirtualSpoutIdentifier identifier,
+        final ConsumerState startingState,
+        final ConsumerState endingState
+    );
+}

src/main/java/com/salesforce/storm/spout/dynamic/DynamicSpout.java

@@ -92,6 +92,11 @@ public class DynamicSpout extends BaseRichSpout {
      */
     private final FactoryManager factoryManager;
 
+    /**
+     * Factory creating {@link DelegateSpout} instances.
+     */
+    private DelegateSpoutFactory virtualSpoutFactory;
+
     /**
      * Handler for callbacks at various stages of a dynamic spout's lifecycle.
      */
@@ -125,7 +130,7 @@ public DynamicSpout(Map<String, Object> spoutConfig) {
         this.spoutConfig = Collections.unmodifiableMap(SpoutConfig.setDefaults(spoutConfig));
 
         // Create our factory manager, which must be serializable.
-        factoryManager = new FactoryManager(getSpoutConfig());
+        this.factoryManager = new FactoryManager(getSpoutConfig());
     }
 
     /**
@@ -157,8 +162,8 @@ public void open(Map topologyConfig, TopologyContext topologyContext, SpoutOutpu
         // finished setting all of these things up.
 
         // Initialize Metrics Collection
-        metricsRecorder = getFactoryManager().createNewMetricsRecorder();
-        metricsRecorder.open(getSpoutConfig(), getTopologyContext());
+        this.metricsRecorder = getFactoryManager().createNewMetricsRecorder();
+        this.metricsRecorder.open(getSpoutConfig(), getTopologyContext());
 
         // Create MessageBuffer
         final MessageBuffer messageBuffer = getFactoryManager().createNewMessageBufferInstance();
@@ -186,14 +191,23 @@ public void open(Map topologyConfig, TopologyContext topologyContext, SpoutOutpu
         // For emit metrics
         emitCountMetrics = Maps.newHashMap();
 
+        // TODO: This should be configurable and created dynamically, the problem is that right now we are still tightly
+        // coupled to the VirtualSpout implementation.
+        this.virtualSpoutFactory = new VirtualSpoutFactory(
+            spoutConfig,
+            topologyContext,
+            factoryManager,
+            metricsRecorder
+        );
+
         // Our spout is open, it's not dependent upon the handler to finish opening for us to be 'opened'
         // This is important, because if we waited most of our getters that check the opened state of the
         // spout would throw an exception and make them unusable.
         isOpen = true;
 
-        spoutHandler = getFactoryManager().createSpoutHandler();
-        spoutHandler.open(spoutConfig);
-        spoutHandler.onSpoutOpen(this, topologyConfig, topologyContext);
+        this.spoutHandler = getFactoryManager().createSpoutHandler();
+        this.spoutHandler.open(spoutConfig, virtualSpoutFactory);
+        this.spoutHandler.onSpoutOpen(this, topologyConfig, topologyContext);
     }
 
     /**

src/main/java/com/salesforce/storm/spout/dynamic/FactoryManager.java

@@ -25,6 +25,7 @@
 
 package com.salesforce.storm.spout.dynamic;
 
+import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
 import com.salesforce.storm.spout.dynamic.config.SpoutConfig;
 import com.salesforce.storm.spout.dynamic.consumer.Consumer;
@@ -66,44 +67,44 @@ public FactoryManager(Map<String, Object> spoutConfig) {
      * @return returns a new instance of the configured RetryManager.
      */
     public RetryManager createNewFailedMsgRetryManagerInstance() {
-        return createNewInstance(
-            (String) spoutConfig.get(SpoutConfig.RETRY_MANAGER_CLASS)
+        return createNewInstanceFromConfig(
+            SpoutConfig.RETRY_MANAGER_CLASS
         );
     }
 
     /**
      * @return returns a new instance of the configured persistence manager.
      */
     public PersistenceAdapter createNewPersistenceAdapterInstance() {
-        return createNewInstance(
-            (String) spoutConfig.get(SpoutConfig.PERSISTENCE_ADAPTER_CLASS)
+        return createNewInstanceFromConfig(
+            SpoutConfig.PERSISTENCE_ADAPTER_CLASS
         );
     }
 
     /**
      * @return returns a new instance of the configured Metrics Recorder manager.
      */
     public MetricsRecorder createNewMetricsRecorder() {
-        return createNewInstance(
-            (String) spoutConfig.get(SpoutConfig.METRICS_RECORDER_CLASS)
+        return createNewInstanceFromConfig(
+            SpoutConfig.METRICS_RECORDER_CLASS
         );
     }
 
     /**
      * @return returns a new instance of the configured MessageBuffer interface.
      */
     public MessageBuffer createNewMessageBufferInstance() {
-        return createNewInstance(
-            (String) spoutConfig.get(SpoutConfig.TUPLE_BUFFER_CLASS)
+        return createNewInstanceFromConfig(
+            SpoutConfig.TUPLE_BUFFER_CLASS
         );
     }
 
     /**
      * @return returns a new instance of the configured Consumer interface.
      */
     public Consumer createNewConsumerInstance() {
-        return createNewInstance(
-            (String) spoutConfig.get(SpoutConfig.CONSUMER_CLASS)
+        return createNewInstanceFromConfig(
+            SpoutConfig.CONSUMER_CLASS
         );
     }
 
@@ -112,8 +113,8 @@ public Consumer createNewConsumerInstance() {
      * @return Instance of a SpoutHandler
      */
     public SpoutHandler createSpoutHandler() {
-        return createNewInstance(
-            (String) spoutConfig.get(SpoutConfig.SPOUT_HANDLER_CLASS)
+        return createNewInstanceFromConfig(
+            SpoutConfig.SPOUT_HANDLER_CLASS
         );
     }
 
@@ -122,8 +123,19 @@ public SpoutHandler createSpoutHandler() {
      * @return Instance of a VirtualSpoutHandler
      */
     public synchronized VirtualSpoutHandler createVirtualSpoutHandler() {
+        return createNewInstanceFromConfig(
+            SpoutConfig.VIRTUAL_SPOUT_HANDLER_CLASS
+        );
+    }
+
+    private synchronized <T> T createNewInstanceFromConfig(final String configKey) {
+        Preconditions.checkState(
+            spoutConfig.containsKey(configKey),
+            "Class has not been specified for configuration key " + configKey
+        );
+
         return createNewInstance(
-            (String) spoutConfig.get(SpoutConfig.VIRTUAL_SPOUT_HANDLER_CLASS)
+            (String) spoutConfig.get(configKey)
         );
     }
 

src/main/java/com/salesforce/storm/spout/dynamic/VirtualSpoutFactory.java

@@ -0,0 +1,126 @@
+/**
+ * Copyright (c) 2017, Salesforce.com, Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without modification, are permitted provided that the
+ * following conditions are met:
+ *
+ * * Redistributions of source code must retain the above copyright notice, this list of conditions and the following
+ *   disclaimer.
+ *
+ * * Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following
+ *   disclaimer in the documentation and/or other materials provided with the distribution.
+ *
+ * * Neither the name of Salesforce.com nor the names of its contributors may be used to endorse or promote products
+ *   derived from this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
+ * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+ * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+ * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
+ * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
+ * USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package com.salesforce.storm.spout.dynamic;
+
+import com.salesforce.storm.spout.dynamic.consumer.ConsumerState;
+import com.salesforce.storm.spout.dynamic.metrics.MetricsRecorder;
+import org.apache.storm.task.TopologyContext;
+
+import java.util.Map;
+
+/**
+ * Factory for easily creating {@link DelegateSpout} instances.
+ *
+ * Handy in things like {@link com.salesforce.storm.spout.dynamic.handler.SpoutHandler} where we want to abstract away particulars, such
+ * as the things passed down from {@link DynamicSpout} such as the {@link MetricsRecorder} as an example.
+ */
+public class VirtualSpoutFactory implements DelegateSpoutFactory {
+
+    /**
+     * Spout configuration.
+     */
+    private final Map<String, Object> spoutConfig;
+    /**
+     * Topology context.
+     */
+    private final TopologyContext topologyContext;
+    /**
+     * Factory manager, for creating instances of classes driven by configuration.
+     */
+    private final FactoryManager factoryManager;
+    /**
+     * Metrics recorder, for capturing metrics.
+     */
+    private final MetricsRecorder metricsRecorder;
+
+    /**
+     * Factory for easily creating {@link DelegateSpout} instances.
+     *
+     * Handy in things like {@link com.salesforce.storm.spout.dynamic.handler.SpoutHandler} where we want to abstract away particulars, such
+     * as the things passed down from {@link DynamicSpout} such as the {@link MetricsRecorder} as an example.
+     */
+    public VirtualSpoutFactory(
+        final Map<String, Object> spoutConfig,
+        final TopologyContext topologyContext,
+        final FactoryManager factoryManager,
+        final MetricsRecorder metricsRecorder
+    ) {
+        this.spoutConfig = spoutConfig;
+        this.topologyContext = topologyContext;
+        this.factoryManager = factoryManager;
+        this.metricsRecorder = metricsRecorder;
+    }
+
+    /**
+     * Create a {@link DelegateSpout} instance.
+     * @param identifier identifier to use for this instance.
+     * @param startingState starting consumer state for this instance.
+     * @param endingState ending consumer state for this instance.
+     * @return instance of {@link DelegateSpout}.
+     */
+    @Override
+    public VirtualSpout create(
+        final VirtualSpoutIdentifier identifier,
+        final ConsumerState startingState,
+        final ConsumerState endingState
+    ) {
+        return new VirtualSpout(
+            identifier,
+            this.spoutConfig,
+            this.topologyContext,
+            this.factoryManager,
+            this.metricsRecorder,
+            startingState,
+            endingState
+        );
+    }
+
+    /**
+     * Create a {@link DelegateSpout} instance.
+     * @param identifier identifier to use for this instance.
+     * @param startingState starting consumer state for this instance.
+     * @return instance of {@link DelegateSpout}.
+     */
+    @Override
+    public VirtualSpout create(
+        final VirtualSpoutIdentifier identifier,
+        final ConsumerState startingState
+    ) {
+        return create(identifier, startingState, null);
+    }
+
+    /**
+     * Create a {@link DelegateSpout} instance.
+     * @param identifier identifier to use for this instance.
+     * @return instance of {@link DelegateSpout}.
+     */
+    @Override
+    public VirtualSpout create(
+        final VirtualSpoutIdentifier identifier
+    ) {
+        return create(identifier, null, null);
+    }
+}

src/main/java/com/salesforce/storm/spout/dynamic/config/SpoutConfig.java

@@ -26,6 +26,7 @@
 package com.salesforce.storm.spout.dynamic.config;
 
 import com.google.common.collect.Maps;
+import com.salesforce.storm.spout.dynamic.VirtualSpoutFactory;
 import com.salesforce.storm.spout.dynamic.config.annotation.Documentation;
 import com.salesforce.storm.spout.dynamic.handler.NoopSpoutHandler;
 import com.salesforce.storm.spout.dynamic.handler.NoopVirtualSpoutHandler;
@@ -393,6 +394,13 @@ public class SpoutConfig {
     )
     public static final String VIRTUAL_SPOUT_HANDLER_CLASS = "spout.virtual_spout_handler_class";
 
+    @Documentation(
+        description = "Defines which DelegateSpoutFactory implementation to use. "
+        + "Should be a fully qualified class path that implements the DelegateSpoutFactory interface.",
+        type = String.class
+    )
+    public static final String VIRTUAL_SPOUT_FACTORY_CLASS = "spout.virtual_spout_factory_class";
+
     /**
      * Logger for logging logs.
      */
@@ -571,6 +579,15 @@ public static Map<String, Object> setDefaults(Map config) {
             );
         }
 
+        if (!clonedConfig.containsKey(VIRTUAL_SPOUT_FACTORY_CLASS)) {
+            clonedConfig.put(VIRTUAL_SPOUT_FACTORY_CLASS, VirtualSpoutFactory.class);
+            logger.info(
+                "Unspecified configuration value for {} using default value {}",
+                VIRTUAL_SPOUT_FACTORY_CLASS,
+                clonedConfig.get(VIRTUAL_SPOUT_FACTORY_CLASS)
+            );
+        }
+
         return clonedConfig;
     }
 }