Merge pull request tomdz#3 from maxichan/master

Bolt updates - pass class instead of strings for input/output fields, fixed tests

pom.xml

@@ -9,7 +9,7 @@
   <groupId>org.tomdz.storm</groupId>
   <artifactId>storm-esper</artifactId>
   <name>storm-esper</name>
-  <version>1.0.2</version>
+  <version>1.0.4</version>
   <packaging>jar</packaging>
   <description>Storm + Esper integration</description>
   <licenses>
@@ -67,12 +67,16 @@
        </snapshots>
     </repository>
   </repositories>
-
   <dependencies>
     <dependency>
       <groupId>org.apache.storm</groupId>
       <artifactId>storm-core</artifactId>
       <version>1.0.2</version>
+      <!--
+        Use "provided" scope to keep storm out of the jar-with-dependencies
+        For IntelliJ dev, intellij will load properly.
+      -->
+      <scope>${provided.scope}</scope>
     </dependency>
     <dependency>
       <groupId>com.espertech</groupId>
@@ -260,5 +264,11 @@
         </plugins>
       </build>
     </profile>
+    <profile>
+      <id>intellij</id>
+      <properties>
+        <provided.scope>compile</provided.scope>
+      </properties>
+    </profile>
   </profiles>
 </project>

src/main/java/org/tomdz/storm/esper/EsperBolt.java

@@ -9,6 +9,7 @@
 import org.apache.storm.topology.base.BaseRichBolt;
 import org.apache.storm.tuple.Fields;
 import org.apache.storm.tuple.Tuple;
+import org.tomdz.storm.esper.model.Event;
 
 import java.util.*;
 
@@ -74,6 +75,28 @@ public AliasedInputBuilder aliasStream(String componentId, String streamId)
         {
             return new AliasedInputBuilder(bolt, new StreamId(componentId, streamId));
         }
+
+        public InputsBuilder onStreamEvent(Event event, String componentId)
+        {
+            return getInputsBuilderFromAliasedBuilder(
+                aliasStream(componentId, event.getStreamName()),
+                event
+            );
+        }
+
+        public InputsBuilder onComponentEvent(Event event, String componentId)
+        {
+            return getInputsBuilderFromAliasedBuilder(aliasComponent(componentId), event);
+        }
+
+        private InputsBuilder getInputsBuilderFromAliasedBuilder(AliasedInputBuilder aliasedBuilder, Event event)
+        {
+            Map<Class, String[]> model = event.getModelByType();
+            for (Class field : model.keySet()) {
+                aliasedBuilder = aliasedBuilder.withFields(model.get(field)).ofType(field);
+            }
+            return aliasedBuilder.toEventType(event.getEventName());
+        }
     }
 
     public static final class AliasedInputBuilder
@@ -151,6 +174,11 @@ public OutputStreamBuilder onDefaultStream()
         {
             return new OutputStreamBuilder(bolt, "default");
         }
+
+        public OutputsBuilder onEvent(Event event)
+        {
+            return onStream(event.getStreamName()).fromEventType(event.getEventName()).emit(event.getModels());
+        }
     }
 
     public static final class OutputStreamBuilder

src/main/java/org/tomdz/storm/esper/model/Event.java

@@ -0,0 +1,10 @@
+package org.tomdz.storm.esper.model;
+
+import java.util.Map;
+
+public interface Event {
+    Map<Class, String[]> getModelByType();
+    String[] getModels();
+    String getStreamName();
+    String getEventName();
+}

src/test/java/org/tomdz/storm/esper/EsperTestJob.java

@@ -0,0 +1,52 @@
+package org.tomdz.storm.esper;
+
+import org.apache.storm.Config;
+import org.apache.storm.testing.CompleteTopologyParam;
+import org.apache.storm.testing.MockedSources;
+import org.apache.storm.testing.TestJob;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.Values;
+
+import java.util.HashSet;
+
+import static org.testng.Assert.assertEquals;
+
+public abstract class EsperTestJob implements TestJob {
+
+    /**
+     * Check that all outgoing fields from the Esper bolt match the incoming fields of the
+     * gathering bolt.
+     * @param esperBolt
+     * @param gatheringBolt
+     */
+    public void assertEventTypesEqual(EsperBolt esperBolt, GatheringBolt gatheringBolt) {
+        for (Tuple tuple : gatheringBolt.getGatheredData()) {
+            String streamId = tuple.getSourceStreamId();
+
+            EventTypeDescriptor eventType = esperBolt.getEventTypeForStreamId(streamId);
+
+            assertEquals(new HashSet<String>(tuple.getFields().toList()),
+                new HashSet<String>(eventType.getFields().toList()));
+        }
+    }
+
+    public CompleteTopologyParam createTestDataConfig(String spoutName, Values... values) {
+        MockedSources mockedSources = new MockedSources();
+        mockedSources.addMockData(spoutName, values);
+
+        return createConfig(mockedSources);
+    }
+
+    public CompleteTopologyParam createConfig(MockedSources mockedSources) {
+        Config conf = new Config();
+        conf.setNumWorkers(1);
+        conf.setDebug(false);
+        conf.put(Config.TOPOLOGY_DEBUG, false);
+        conf.put(Config.TOPOLOGY_EVENTLOGGER_EXECUTORS, 0);
+
+        CompleteTopologyParam completeTopologyParam = new CompleteTopologyParam();
+        completeTopologyParam.setMockedSources(mockedSources);
+        completeTopologyParam.setStormConf(conf);
+        return completeTopologyParam;
+    }
+}