package backtype.storm.testing;

import backtype.storm.coordination.BatchOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBatchBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;

public class BatchNumberList extends BaseBatchBolt {

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("word", "list"));
    }

    String _wordComponent;
    
    public BatchNumberList(String wordComponent) {
        _wordComponent = wordComponent;
    }
    
    String word = null;
    List<Integer> intSet = new ArrayList<Integer>();
    BatchOutputCollector _collector;
    
    
    @Override
    public void prepare(Map conf, TopologyContext context, BatchOutputCollector collector, Object id) {
        _collector = collector;
        System.out.println("STARTED: " + id);
    }

    @Override
    public void execute(Tuple tuple) {
        System.out.println("RECEIVED: " + tuple.toString());
        if(tuple.getSourceComponent().equals(_wordComponent)) {
            this.word = tuple.getString(1);
        } else {
            intSet.add(tuple.getInteger(1));
        }
    }

    @Override
    public void finishBatch() {
        if(word!=null) {
            Collections.sort(intSet);
            _collector.emit(new Values(word, intSet));
        }
    }
    
}

@@ -1052,54 +1052,8 @@
 or;%0A
-        System.out.println(%22STARTED: %22 + id);%0A
     
@@ -1112,69 +1112,8 @@
 ) %7B%0A
-        System.out.println(%22RECEIVED: %22 + tuple.toString());%0A
     

remove debug prints