test_class.file: storm/trident-examples/src/test/java/storm/trident/topology/WordCountTest.java
test_class.identifier: WordCountTest
test_case.body: @Test
      public void testRunTopology() throws InterruptedException {
          Configuration conf = new Configuration();
          WordCount topology = new WordCount("wordcount", conf);
          
          StormRunner.runTopologyLocally(topology.buildTopology(), "wordcount", conf, 100);
      }
test_case.class_method_signature: WordCountTest.testRunTopology()
test_case.full_signature: @Test public void testRunTopology()
test_case.identifier: testRunTopology
test_case.invocations: runTopologyLocally buildTopology
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testRunTopology()
focal_class.declarator: spoutType
focal_class.modifier: private
focal_class.original_string: private String spoutType;
focal_class.type: String
focal_class.var_name: spoutType
focal_class.declarator: sinkType
focal_class.modifier: private
focal_class.original_string: private String sinkType;
focal_class.type: String
focal_class.var_name: sinkType
focal_class.declarator: spoutThreads
focal_class.modifier: private
focal_class.original_string: private int spoutThreads;
focal_class.type: int
focal_class.var_name: spoutThreads
focal_class.declarator: splitterThreads
focal_class.modifier: private
focal_class.original_string: private int splitterThreads;
focal_class.type: int
focal_class.var_name: splitterThreads
focal_class.declarator: counterThreads
focal_class.modifier: private
focal_class.original_string: private int counterThreads;
focal_class.type: int
focal_class.var_name: counterThreads
focal_class.declarator: sinkThreads
focal_class.modifier: private
focal_class.original_string: private int sinkThreads;
focal_class.type: int
focal_class.var_name: sinkThreads
focal_class.file: storm/trident-examples/src/main/java/storm/trident/topology/WordCount.java
focal_class.identifier: WordCount
focal_class.class_method_signature: WordCount.WordCount(String topologyName, Configuration config)
focal_class.full_signature: public  WordCount(String topologyName, Configuration config)
focal_class.identifier: WordCount
focal_class.modifiers: public
focal_class.parameters: (String topologyName, Configuration config)
focal_class.signature:  WordCount(String topologyName, Configuration config)
focal_class.class_method_signature: WordCount.initialize()
focal_class.full_signature: @Override public void initialize()
focal_class.identifier: initialize
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void initialize()
focal_class.class_method_signature: WordCount.buildTopology()
focal_class.full_signature: @Override public StormTopology buildTopology()
focal_class.identifier: buildTopology
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: StormTopology
focal_class.signature: StormTopology buildTopology()
focal_class.superclass: extends AbstractTopology
focal_method.body: @Override
      public StormTopology buildTopology() {
          TridentTopology topology = new TridentTopology();
          
          Stream sentences = null;
          
          switch (spoutType) {
              case BaseConstants.SourceType.FILE:
                  int maxBatchSize = config.getInt(Config.SPOUT_MAX_BATCH_SIZE, 100);
                  String spoutPath = config.getString(Config.SPOUT_PATH);
                  sentences = topology.newStream("spout", new FileSpout(maxBatchSize, spoutPath));
                  break;
              case BaseConstants.SourceType.KAFKA:
                  createKafkaSpout(spoutType, spoutType);
                  break;
              default:
                  throw new IllegalArgumentException("Must inform a spout type");
          }
          
          sentences = sentences.parallelismHint(spoutThreads);
                  
          GroupedStream words = sentences.each(new Fields("sentence"), new Split(), new Fields("word"))
                  .parallelismHint(splitterThreads)
                  .groupBy(new Fields("word"));
                  
          Stream counts = words.persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count"))
                  .parallelismHint(counterThreads)
                  .newValuesStream();
          
          if (sinkType.equals(SinkType.CONSOLE)) {
              counts.each(new Fields("word", "count"), new Debug())
                      .parallelismHint(sinkThreads);
          }
  
          return topology.build();
      }
focal_method.class_method_signature: WordCount.buildTopology()
focal_method.full_signature: @Override public StormTopology buildTopology()
focal_method.identifier: buildTopology
focal_method.invocations: getInt getString newStream createKafkaSpout parallelismHint groupBy parallelismHint each newValuesStream parallelismHint persistentAggregate equals parallelismHint each build
focal_method.modifiers: @Override public
focal_method.parameters: ()
focal_method.return: StormTopology
focal_method.signature: StormTopology buildTopology()