// 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/RegexFilter.java
// public class RegexFilter extends BaseFilter {
//     private final Pattern pattern;
// 
//     public RegexFilter(String pattern) {
//         this.pattern = Pattern.compile(pattern);
//     }
// 
//     @Override
//     public boolean isKeep(TridentTuple tuple) {
// 
//         String string = tuple.getString(0);
//         return pattern.matcher(string).matches();
//     }
// }
// 
// Path: src/main/java/tutorial/storm/trident/operations/ToUpperCase.java
// @SuppressWarnings("serial")
// public class ToUpperCase extends BaseFunction {
// 
//     @Override
//     public void execute(TridentTuple tuple, TridentCollector collector) {
//         collector.emit(new Values(tuple.getString(0).toUpperCase()));
//     }
// }
// 
// 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 org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import storm.trident.TridentTopology;
import storm.trident.operation.builtin.Count;
import storm.trident.spout.IBatchSpout;
import storm.trident.testing.MemoryMapState;
import tutorial.storm.trident.operations.Print;
import tutorial.storm.trident.operations.RegexFilter;
import tutorial.storm.trident.operations.ToUpperCase;
import tutorial.storm.trident.testutil.FakeTweetsBatchSpout;
import java.io.IOException;
package tutorial.storm.trident;




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

    public static void main(String[] args) throws Exception {
        // Storm can be run locally for testing purposes
        Config conf = new Config();
//        conf.put(Config.TOPOLOGY_DEBUG,true);

        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("basic_primitives", conf, basicPrimitives(new FakeTweetsBatchSpout()));
        Thread.sleep(100000);
    }

    public static StormTopology basicPrimitives(IBatchSpout spout) throws IOException {

        // A topology is a set of streams.
        // A stream is a DAG of Spouts and Bolts.
        // (In Storm there are Spouts (data producers) and Bolts (data processors).
        // Spouts create Tuples and Bolts manipulate then and possibly emit new ones.)

        // But in Trident we operate at a higher level.
        // Bolts are created and connected automatically out of higher-level constructs.
        // Also, Spouts are "batched".
        TridentTopology topology = new TridentTopology();

        // The "each" primitive allows us to apply either filters or functions to the stream
        // We always have to select the input fields.
        topology
                .newStream("filter", spout)
                .each(new Fields("actor"), new RegexFilter("pere"))
                .each(new Fields("text", "actor"), new Print());

        // Functions describe their output fields, which are always appended to the input fields.
        // As you see, Each operations can be chained.
        topology
                .newStream("function", spout)
                .each(new Fields("text"), new ToUpperCase(), new Fields("uppercased_text"))
                .each(new Fields("text", "uppercased_text"), new Print());

        // You can prune unnecessary fields using "project"
        topology
                .newStream("projection", spout)
                .each(new Fields("text"), new ToUpperCase(), new Fields("uppercased_text"))
                .project(new Fields("uppercased_text"))
                .each(new Fields("uppercased_text"), new Print());

        // Stream can be parallelized with "parallelismHint"
        // Parallelism hint is applied downwards until a partitioning operation (we will see this later).
        // This topology creates 5 spouts and 5 bolts:
        // Let's debug that with TridentOperationContext.partitionIndex !
// 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/RegexFilter.java
// public class RegexFilter extends BaseFilter {
//     private final Pattern pattern;
// 
//     public RegexFilter(String pattern) {
//         this.pattern = Pattern.compile(pattern);
//     }
// 
//     @Override
//     public boolean isKeep(TridentTuple tuple) {
// 
//         String string = tuple.getString(0);
//         return pattern.matcher(string).matches();
//     }
// }
// 
// Path: src/main/java/tutorial/storm/trident/operations/ToUpperCase.java
// @SuppressWarnings("serial")
// public class ToUpperCase extends BaseFunction {
// 
//     @Override
//     public void execute(TridentTuple tuple, TridentCollector collector) {
//         collector.emit(new Values(tuple.getString(0).toUpperCase()));
//     }
// }
// 
// 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/Part01_BasicPrimitives.java
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.generated.StormTopology;
import backtype.storm.tuple.Fields;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import storm.trident.TridentTopology;
import storm.trident.operation.builtin.Count;
import storm.trident.spout.IBatchSpout;
import storm.trident.testing.MemoryMapState;
import tutorial.storm.trident.operations.Print;
import tutorial.storm.trident.operations.RegexFilter;
import tutorial.storm.trident.operations.ToUpperCase;
import tutorial.storm.trident.testutil.FakeTweetsBatchSpout;
import java.io.IOException;

package tutorial.storm.trident;




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

    public static void main(String[] args) throws Exception {
        // Storm can be run locally for testing purposes
        Config conf = new Config();
//        conf.put(Config.TOPOLOGY_DEBUG,true);

        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("basic_primitives", conf, basicPrimitives(new FakeTweetsBatchSpout()));
        Thread.sleep(100000);
    }

    public static StormTopology basicPrimitives(IBatchSpout spout) throws IOException {

        // A topology is a set of streams.
        // A stream is a DAG of Spouts and Bolts.
        // (In Storm there are Spouts (data producers) and Bolts (data processors).
        // Spouts create Tuples and Bolts manipulate then and possibly emit new ones.)

        // But in Trident we operate at a higher level.
        // Bolts are created and connected automatically out of higher-level constructs.
        // Also, Spouts are "batched".
        TridentTopology topology = new TridentTopology();

        // The "each" primitive allows us to apply either filters or functions to the stream
        // We always have to select the input fields.
        topology
                .newStream("filter", spout)
                .each(new Fields("actor"), new RegexFilter("pere"))
                .each(new Fields("text", "actor"), new Print());

        // Functions describe their output fields, which are always appended to the input fields.
        // As you see, Each operations can be chained.
        topology
                .newStream("function", spout)
                .each(new Fields("text"), new ToUpperCase(), new Fields("uppercased_text"))
                .each(new Fields("text", "uppercased_text"), new Print());

        // You can prune unnecessary fields using "project"
        topology
                .newStream("projection", spout)
                .each(new Fields("text"), new ToUpperCase(), new Fields("uppercased_text"))
                .project(new Fields("uppercased_text"))
                .each(new Fields("uppercased_text"), new Print());

        // Stream can be parallelized with "parallelismHint"
        // Parallelism hint is applied downwards until a partitioning operation (we will see this later).
        // This topology creates 5 spouts and 5 bolts:
        // Let's debug that with TridentOperationContext.partitionIndex !
        topology