package spouts;

import java.util.Map;

import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.tuple.Values;
import backtype.storm.tuple.Fields;

public class ItemGenerator extends BaseRichSpout {
  private SpoutOutputCollector _collector;

  @Override
  public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
    _collector = collector;
  }

  @Override
  public void nextTuple() {
    _collector.emit(new Values("Hello storm"));
  }

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

@@ -397,16 +397,53 @@
 llector;
+%0A  private boolean completed = false;
 %0A%0A  @Ove
@@ -605,24 +605,53 @@
 xtTuple() %7B%0A
+    if(false == completed)%0A  
     _collect
@@ -682,24 +682,46 @@
 o storm%22));%0A
+    completed = true;%0A
   %7D%0A%0A  @Over

emit one message