// Path: src/main/java/tutorial/storm/trident/operations/Print.java
// public class Print extends BaseFilter {
//     private int partitionIndex;
//     private int numPartitions;
//     private final String name;
// 
//     public Print(){
//         name = "";
//     }
//     public Print(String name){
//         this.name = name;
//     }
// 
//     @Override
//     public void prepare(Map conf, TridentOperationContext context) {
//         this.partitionIndex = context.getPartitionIndex();
//         this.numPartitions = context.numPartitions();
//     }
// 
// 
//     @Override
//     public boolean isKeep(TridentTuple tuple) {
//         System.err.println(String.format("%s::Partition idx: %s out of %s partitions got %s", name, partitionIndex, numPartitions, tuple.toString()));
//         return true;
//     }
// }
// 
// Path: src/main/java/tutorial/storm/trident/operations/StringCounter.java
// public class StringCounter implements Aggregator<Map<String, Integer>> {
// 
//     private int partitionId;
//     private int numPartitions;
// 
//     @Override
//     public void prepare(Map conf, TridentOperationContext context) {
//         this.partitionId = context.getPartitionIndex();
//         this.numPartitions = context.numPartitions();
//     }
// 
//     @Override
//     public void cleanup() {
//     }
// 
//     @Override
//     public Map<String, Integer> init(Object batchId, TridentCollector collector) {
//         return new HashMap<String, Integer>();
//     }
// 
//     @Override
//     public void aggregate(Map<String, Integer> val, TridentTuple tuple, TridentCollector collector) {
//         String loc = tuple.getString(0);
//         Integer previousValue = val.get(loc);
//         previousValue = previousValue == null ? 0 : previousValue;
//         val.put(loc, previousValue + 1);
//     }
// 
//     @Override
//     public void complete(Map<String, Integer> val, TridentCollector collector) {
//         System.err.println(String.format("Partition %s out ot %s partitions aggregated:%s", partitionId, numPartitions, val));
//         collector.emit(new Values(val));
//     }
// }
// 
// 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.LocalCluster;
import backtype.storm.generated.StormTopology;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import storm.trident.TridentTopology;
import storm.trident.operation.BaseFilter;
import storm.trident.operation.BaseFunction;
import storm.trident.operation.TridentCollector;
import storm.trident.tuple.TridentTuple;
import tutorial.storm.trident.operations.Print;
import tutorial.storm.trident.operations.StringCounter;
import tutorial.storm.trident.testutil.FakeTweetsBatchSpout;
import java.util.HashMap;
import java.util.Map;
package tutorial.storm.trident;




/**
 * @author Enno Shioji (enno.shioji@peerindex.com)
 */
public class Part02_AdvancedPrimitives1 {
    private static final Logger log = LoggerFactory.getLogger(Part02_AdvancedPrimitives1.class);

    public static void main(String[] args) throws Exception {
        Config conf = new Config();
//        conf.put(Config.TOPOLOGY_DEBUG,true);
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("advanced_primitives", conf, advancedPrimitives(new FakeTweetsBatchSpout(1000)));
        Thread.sleep(30000);
        cluster.shutdown();
    }

// Path: src/main/java/tutorial/storm/trident/operations/Print.java
// public class Print extends BaseFilter {
//     private int partitionIndex;
//     private int numPartitions;
//     private final String name;
// 
//     public Print(){
//         name = "";
//     }
//     public Print(String name){
//         this.name = name;
//     }
// 
//     @Override
//     public void prepare(Map conf, TridentOperationContext context) {
//         this.partitionIndex = context.getPartitionIndex();
//         this.numPartitions = context.numPartitions();
//     }
// 
// 
//     @Override
//     public boolean isKeep(TridentTuple tuple) {
//         System.err.println(String.format("%s::Partition idx: %s out of %s partitions got %s", name, partitionIndex, numPartitions, tuple.toString()));
//         return true;
//     }
// }
// 
// Path: src/main/java/tutorial/storm/trident/operations/StringCounter.java
// public class StringCounter implements Aggregator<Map<String, Integer>> {
// 
//     private int partitionId;
//     private int numPartitions;
// 
//     @Override
//     public void prepare(Map conf, TridentOperationContext context) {
//         this.partitionId = context.getPartitionIndex();
//         this.numPartitions = context.numPartitions();
//     }
// 
//     @Override
//     public void cleanup() {
//     }
// 
//     @Override
//     public Map<String, Integer> init(Object batchId, TridentCollector collector) {
//         return new HashMap<String, Integer>();
//     }
// 
//     @Override
//     public void aggregate(Map<String, Integer> val, TridentTuple tuple, TridentCollector collector) {
//         String loc = tuple.getString(0);
//         Integer previousValue = val.get(loc);
//         previousValue = previousValue == null ? 0 : previousValue;
//         val.put(loc, previousValue + 1);
//     }
// 
//     @Override
//     public void complete(Map<String, Integer> val, TridentCollector collector) {
//         System.err.println(String.format("Partition %s out ot %s partitions aggregated:%s", partitionId, numPartitions, val));
//         collector.emit(new Values(val));
//     }
// }
// 
// 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/Part02_AdvancedPrimitives1.java
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.generated.StormTopology;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import storm.trident.TridentTopology;
import storm.trident.operation.BaseFilter;
import storm.trident.operation.BaseFunction;
import storm.trident.operation.TridentCollector;
import storm.trident.tuple.TridentTuple;
import tutorial.storm.trident.operations.Print;
import tutorial.storm.trident.operations.StringCounter;
import tutorial.storm.trident.testutil.FakeTweetsBatchSpout;
import java.util.HashMap;
import java.util.Map;

package tutorial.storm.trident;




/**
 * @author Enno Shioji (enno.shioji@peerindex.com)
 */
public class Part02_AdvancedPrimitives1 {
    private static final Logger log = LoggerFactory.getLogger(Part02_AdvancedPrimitives1.class);

    public static void main(String[] args) throws Exception {
        Config conf = new Config();
//        conf.put(Config.TOPOLOGY_DEBUG,true);
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("advanced_primitives", conf, advancedPrimitives(new FakeTweetsBatchSpout(1000)));
        Thread.sleep(30000);
        cluster.shutdown();
    }

    private static StormTopology advancedPrimitives(FakeTweetsBatchSpout spout) {