package storm.trident.testing;

import backtype.storm.Config;
import backtype.storm.task.TopologyContext;
import backtype.storm.tuple.Fields;
import java.util.List;
import java.util.Map;
import storm.trident.operation.TridentCollector;
import storm.trident.spout.IBatchSpout;


public class FixedBatchSpout implements IBatchSpout {

    Fields fields;
    List<Object>[] outputs;
    int maxBatchSize;
    
    public FixedBatchSpout(Fields fields, int maxBatchSize, List<Object>... outputs) {
        this.fields = fields;
        this.outputs = outputs;
        this.maxBatchSize = maxBatchSize;
    }
    
    int index = 0;
    boolean cycle = false;
    
    public void setCycle(boolean cycle) {
        this.cycle = cycle;
    }
    
    @Override
    public void open(Map conf, TopologyContext context) {
        index = 0;
    }

    @Override
    public void emitBatch(long batchId, TridentCollector collector) {
        //Utils.sleep(2000);
        if(index>=outputs.length && cycle) {
            index = 0;
        }
        for(int i=0; index < outputs.length && i < maxBatchSize; index++, i++) {
            collector.emit(outputs[index]);
        }
    }

    @Override
    public void ack(long batchId) {
        
    }

    @Override
    public void close() {
    }

    @Override
    public Map getComponentConfiguration() {
        Config conf = new Config();
        conf.setMaxTaskParallelism(1);
        return conf;
    }

    @Override
    public Fields getOutputFields() {
        return fields;
    }
    
}

@@ -134,16 +134,45 @@
 .Fields;
+%0A%0Aimport java.util.ArrayList;
 %0Aimport 
@@ -209,16 +209,43 @@
 il.Map;%0A
+import java.util.HashMap;%0A%0A
 import s
@@ -447,24 +447,113 @@
 xBatchSize;%0A
+    HashMap%3CLong, List%3CList%3CObject%3E%3E%3E batches = new HashMap%3CLong, List%3CList%3CObject%3E%3E%3E();%0A
     %0A    pub
@@ -1073,29 +1073,144 @@
     
-//Utils.sleep(2000);%0A
+List%3CList%3CObject%3E%3E batch = this.batches.get(batchId);%0A        if(batch == null)%7B%0A            batch = new ArrayList%3CList%3CObject%3E%3E();%0A    
     
@@ -1242,32 +1242,36 @@
 gth && cycle) %7B%0A
+    
             inde
@@ -1281,26 +1281,34 @@
  0;%0A        
-%7D%0A
+    %7D%0A    
         for(
@@ -1392,22 +1392,21 @@
     
-collector.emit
+    batch.add
 (out
@@ -1419,16 +1419,160 @@
 ndex%5D);%0A
+            %7D%0A            this.batches.put(batchId, batch);%0A        %7D%0A        for(List%3CObject%3E list : batch)%7B%0A            collector.emit(list);%0A
         
@@ -1634,24 +1634,53 @@
 ) %7B%0A        
+this.batches.remove(batchId);
 %0A    %7D%0A%0A    
@@ -1971,9 +1971,8 @@
 %7D%0A    %0A%7D
-%0A

make FixedBatchSpout batch aware.