// Path: src/main/java/tutorial/storm/trident/testutil/FakeTweetsBatchSpout.java
// @SuppressWarnings({"serial", "rawtypes"})
// public class FakeTweetsBatchSpout implements IBatchSpout {
//     private static final AtomicInteger seq = new AtomicInteger();
// 
//     private int batchSize;
//     private FakeTweetGenerator fakeTweetGenerator;
// 
// 
//     public FakeTweetsBatchSpout() throws IOException {
//         this(5);
//     }
// 
//     public FakeTweetsBatchSpout(int batchSize) throws IOException {
//         this.batchSize = batchSize;
//     }
// 
//     @SuppressWarnings("unchecked")
//     @Override
//     public void open(Map conf, TopologyContext context) {
//         // init
//         System.err.println(this.getClass().getSimpleName() +":" + seq.incrementAndGet()+" opened.");
//         fakeTweetGenerator = new FakeTweetGenerator();
// 
//     }
// 
//     @Override
//     public void emitBatch(long batchId, TridentCollector collector) {
//         // emit batchSize fake tweets
//         for (int i = 0; i < batchSize; i++) {
//             collector.emit(fakeTweetGenerator.getNextTweet());
//         }
//     }
// 
//     @Override
//     public void ack(long batchId) {
//         // nothing to do here
//     }
// 
//     @Override
//     public void close() {
//         // nothing to do here
//     }
// 
//     @Override
//     public Map getComponentConfiguration() {
//         // no particular configuration here
//         return new Config();
//     }
// 
//     @Override
//     public Fields getOutputFields() {
//         return new Fields("id", "text", "actor", "location", "date");
//     }
// }


import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.tuple.Fields;
import storm.trident.TridentState;
import storm.trident.TridentTopology;
import storm.trident.operation.builtin.Count;
import storm.trident.operation.builtin.MapGet;
import storm.trident.testing.MemoryMapState;
import tutorial.storm.trident.testutil.FakeTweetsBatchSpout;
package tutorial.storm.trident;


/**
 * Silly test topology to test the cluster
 *
 * @author Enno Shioji (eshioji@gmail.com)
 */
public class ClusterTestTopology {
    public static void main(String[] args) throws Exception {
        Config conf = new Config();

        // Submits the topology
        String topologyName = args[0];
        conf.setNumWorkers(8); // Our Vagrant environment has 8 workers

// Path: src/main/java/tutorial/storm/trident/testutil/FakeTweetsBatchSpout.java
// @SuppressWarnings({"serial", "rawtypes"})
// public class FakeTweetsBatchSpout implements IBatchSpout {
//     private static final AtomicInteger seq = new AtomicInteger();
// 
//     private int batchSize;
//     private FakeTweetGenerator fakeTweetGenerator;
// 
// 
//     public FakeTweetsBatchSpout() throws IOException {
//         this(5);
//     }
// 
//     public FakeTweetsBatchSpout(int batchSize) throws IOException {
//         this.batchSize = batchSize;
//     }
// 
//     @SuppressWarnings("unchecked")
//     @Override
//     public void open(Map conf, TopologyContext context) {
//         // init
//         System.err.println(this.getClass().getSimpleName() +":" + seq.incrementAndGet()+" opened.");
//         fakeTweetGenerator = new FakeTweetGenerator();
// 
//     }
// 
//     @Override
//     public void emitBatch(long batchId, TridentCollector collector) {
//         // emit batchSize fake tweets
//         for (int i = 0; i < batchSize; i++) {
//             collector.emit(fakeTweetGenerator.getNextTweet());
//         }
//     }
// 
//     @Override
//     public void ack(long batchId) {
//         // nothing to do here
//     }
// 
//     @Override
//     public void close() {
//         // nothing to do here
//     }
// 
//     @Override
//     public Map getComponentConfiguration() {
//         // no particular configuration here
//         return new Config();
//     }
// 
//     @Override
//     public Fields getOutputFields() {
//         return new Fields("id", "text", "actor", "location", "date");
//     }
// }


// Path: src/main/java/tutorial/storm/trident/ClusterTestTopology.java
import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.tuple.Fields;
import storm.trident.TridentState;
import storm.trident.TridentTopology;
import storm.trident.operation.builtin.Count;
import storm.trident.operation.builtin.MapGet;
import storm.trident.testing.MemoryMapState;
import tutorial.storm.trident.testutil.FakeTweetsBatchSpout;

package tutorial.storm.trident;


/**
 * Silly test topology to test the cluster
 *
 * @author Enno Shioji (eshioji@gmail.com)
 */
public class ClusterTestTopology {
    public static void main(String[] args) throws Exception {
        Config conf = new Config();

        // Submits the topology
        String topologyName = args[0];
        conf.setNumWorkers(8); // Our Vagrant environment has 8 workers

        FakeTweetsBatchSpout fakeTweets = new FakeTweetsBatchSpout(10);