// Path: src/main/java/org/apache/storm/hdfs/common/rotation/MoveFileAction.java
// public class MoveFileAction implements RotationAction {
//     private static final Logger LOG = LoggerFactory.getLogger(MoveFileAction.class);
// 
//     private String destination;
// 
//     public MoveFileAction toDestination(String destDir){
//         destination = destDir;
//         return this;
//     }
// 
//     @Override
//     public void execute(FileSystem fileSystem, Path filePath) throws IOException {
//         Path destPath = new Path(destination, filePath.getName());
//         LOG.info("Moving file {} to {}", filePath, destPath);
//         boolean success = fileSystem.rename(filePath, destPath);
//         return;
//     }
// }
// 
// Path: src/main/java/org/apache/storm/hdfs/trident/rotation/FileRotationPolicy.java
// public interface FileRotationPolicy extends Serializable {
//     /**
//      * Called for every tuple the HdfsBolt executes.
//      *
//      * @param tuple The tuple executed.
//      * @param offset current offset of file being written
//      * @return true if a file rotation should be performed
//      */
//     boolean mark(TridentTuple tuple, long offset);
// 
// 
//     /**
//      * Called after the HdfsBolt rotates a file.
//      *
//      */
//     void reset();
// }
// 
// Path: src/main/java/org/apache/storm/hdfs/trident/rotation/FileSizeRotationPolicy.java
// public class FileSizeRotationPolicy implements FileRotationPolicy {
//     private static final Logger LOG = LoggerFactory.getLogger(FileSizeRotationPolicy.class);
// 
//     public static enum Units {
// 
//         KB((long)Math.pow(2, 10)),
//         MB((long)Math.pow(2, 20)),
//         GB((long)Math.pow(2, 30)),
//         TB((long)Math.pow(2, 40));
// 
//         private long byteCount;
// 
//         private Units(long byteCount){
//             this.byteCount = byteCount;
//         }
// 
//         public long getByteCount(){
//             return byteCount;
//         }
//     }
// 
//     private long maxBytes;
// 
//     private long lastOffset = 0;
//     private long currentBytesWritten = 0;
// 
//     public FileSizeRotationPolicy(float count, Units units){
//         this.maxBytes = (long)(count * units.getByteCount());
//     }
// 
//     @Override
//     public boolean mark(TridentTuple tuple, long offset) {
//         long diff = offset - this.lastOffset;
//         this.currentBytesWritten += diff;
//         this.lastOffset = offset;
//         return this.currentBytesWritten >= this.maxBytes;
//     }
// 
//     @Override
//     public void reset() {
//         this.currentBytesWritten = 0;
//         this.lastOffset = 0;
//     }
// 
// }


import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.StormTopology;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import org.apache.storm.hdfs.common.rotation.MoveFileAction;
import org.apache.storm.hdfs.trident.format.*;
import org.apache.storm.hdfs.trident.rotation.FileRotationPolicy;
import org.apache.storm.hdfs.trident.rotation.FileSizeRotationPolicy;
import storm.trident.Stream;
import storm.trident.TridentState;
import storm.trident.TridentTopology;
import storm.trident.operation.BaseFunction;
import storm.trident.operation.TridentCollector;
import storm.trident.state.StateFactory;
import storm.trident.tuple.TridentTuple;
package org.apache.storm.hdfs.trident;


public class TridentSequenceTopology {

    public static StormTopology buildTopology(String hdfsUrl){
        FixedBatchSpout spout = new FixedBatchSpout(new Fields("sentence", "key"), 1000, new Values("the cow jumped over the moon", 1l),
                new Values("the man went to the store and bought some candy", 2l), new Values("four score and seven years ago", 3l),
                new Values("how many apples can you eat", 4l), new Values("to be or not to be the person", 5l));
        spout.setCycle(true);

        TridentTopology topology = new TridentTopology();
        Stream stream = topology.newStream("spout1", spout);

        Fields hdfsFields = new Fields("sentence", "key");

        FileNameFormat fileNameFormat = new DefaultFileNameFormat()
                .withPath("/trident")
                .withPrefix("trident")
                .withExtension(".seq");

// Path: src/main/java/org/apache/storm/hdfs/common/rotation/MoveFileAction.java
// public class MoveFileAction implements RotationAction {
//     private static final Logger LOG = LoggerFactory.getLogger(MoveFileAction.class);
// 
//     private String destination;
// 
//     public MoveFileAction toDestination(String destDir){
//         destination = destDir;
//         return this;
//     }
// 
//     @Override
//     public void execute(FileSystem fileSystem, Path filePath) throws IOException {
//         Path destPath = new Path(destination, filePath.getName());
//         LOG.info("Moving file {} to {}", filePath, destPath);
//         boolean success = fileSystem.rename(filePath, destPath);
//         return;
//     }
// }
// 
// Path: src/main/java/org/apache/storm/hdfs/trident/rotation/FileRotationPolicy.java
// public interface FileRotationPolicy extends Serializable {
//     /**
//      * Called for every tuple the HdfsBolt executes.
//      *
//      * @param tuple The tuple executed.
//      * @param offset current offset of file being written
//      * @return true if a file rotation should be performed
//      */
//     boolean mark(TridentTuple tuple, long offset);
// 
// 
//     /**
//      * Called after the HdfsBolt rotates a file.
//      *
//      */
//     void reset();
// }
// 
// Path: src/main/java/org/apache/storm/hdfs/trident/rotation/FileSizeRotationPolicy.java
// public class FileSizeRotationPolicy implements FileRotationPolicy {
//     private static final Logger LOG = LoggerFactory.getLogger(FileSizeRotationPolicy.class);
// 
//     public static enum Units {
// 
//         KB((long)Math.pow(2, 10)),
//         MB((long)Math.pow(2, 20)),
//         GB((long)Math.pow(2, 30)),
//         TB((long)Math.pow(2, 40));
// 
//         private long byteCount;
// 
//         private Units(long byteCount){
//             this.byteCount = byteCount;
//         }
// 
//         public long getByteCount(){
//             return byteCount;
//         }
//     }
// 
//     private long maxBytes;
// 
//     private long lastOffset = 0;
//     private long currentBytesWritten = 0;
// 
//     public FileSizeRotationPolicy(float count, Units units){
//         this.maxBytes = (long)(count * units.getByteCount());
//     }
// 
//     @Override
//     public boolean mark(TridentTuple tuple, long offset) {
//         long diff = offset - this.lastOffset;
//         this.currentBytesWritten += diff;
//         this.lastOffset = offset;
//         return this.currentBytesWritten >= this.maxBytes;
//     }
// 
//     @Override
//     public void reset() {
//         this.currentBytesWritten = 0;
//         this.lastOffset = 0;
//     }
// 
// }


// Path: src/test/java/org/apache/storm/hdfs/trident/TridentSequenceTopology.java
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.StormTopology;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import org.apache.storm.hdfs.common.rotation.MoveFileAction;
import org.apache.storm.hdfs.trident.format.*;
import org.apache.storm.hdfs.trident.rotation.FileRotationPolicy;
import org.apache.storm.hdfs.trident.rotation.FileSizeRotationPolicy;
import storm.trident.Stream;
import storm.trident.TridentState;
import storm.trident.TridentTopology;
import storm.trident.operation.BaseFunction;
import storm.trident.operation.TridentCollector;
import storm.trident.state.StateFactory;
import storm.trident.tuple.TridentTuple;

package org.apache.storm.hdfs.trident;


public class TridentSequenceTopology {

    public static StormTopology buildTopology(String hdfsUrl){
        FixedBatchSpout spout = new FixedBatchSpout(new Fields("sentence", "key"), 1000, new Values("the cow jumped over the moon", 1l),
                new Values("the man went to the store and bought some candy", 2l), new Values("four score and seven years ago", 3l),
                new Values("how many apples can you eat", 4l), new Values("to be or not to be the person", 5l));
        spout.setCycle(true);

        TridentTopology topology = new TridentTopology();
        Stream stream = topology.newStream("spout1", spout);

        Fields hdfsFields = new Fields("sentence", "key");

        FileNameFormat fileNameFormat = new DefaultFileNameFormat()
                .withPath("/trident")
                .withPrefix("trident")
                .withExtension(".seq");

        FileRotationPolicy rotationPolicy = new FileSizeRotationPolicy(5.0f, FileSizeRotationPolicy.Units.MB);