Understanding Storm Internals by Examples - Real-time Analytics with Storm and Cassandra - page 53

readFile();
}
//this method is called every 100 ms and it polls the list
//for message which is read off as next tuple and emit
the spout to
//the topology. When queue doesn't have any events, it
reads the
//file again calling the readFile method
@
Override
public void nextTuple() {
Utils.sleep(100);
String fullMsg = (String) qe.poll();
String msg[] = (String[]) null;
if (fullMsg != null) {
msg = (new String(fullMsg)).split("#@#");
_collector.emit(new Values(msg[0]));
System.out.println((new StringBuilder("nextTuple
done ")).append(msg[1]).toString());
} else {
readFile();
}
}
@
Override
public void ack(Object id) {}
@
Override
public void fail(Object id) {}
@
Override
public void declareOutputFields(OutputFieldsDeclarer
declarer) {
declarer.declare(new Fields("word"));