nathanmarz/storm — Issue #66: help me !!check my simple code !tks!!!
Author: wangweichen
Created: 2011-12-06 08:56:27+00:00
State: open

spout:
public class RandomSentenceSpout implements IRichSpout {
SpoutOutputCollector _collector;
Random _rand;
public static Logger LOG = Logger.getLogger(RandomSentenceSpout.class);

public boolean isDistributed() {
return true;
}

public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
_collector = collector;
_rand = new Random();
}

public void nextTuple() {
Utils.sleep(100);
String[] sentences = new String[] {
"the cow jumped over the moon",
"an apple a day keeps the doctor away",
"four score and seven years ago",
"snow white and the seven dwarfs",
"i am at two with nature"};
String sentence = sentences[_rand.nextInt(sentences.length)];
System.out.println(sentence);
_collector.emit(new Values(sentence));
}

public void close() {
}

public void ack(Object id) {
}

public void fail(Object id) {
}

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

}

bolt:
package com.taobao.search.bolt.log;

import java.util.Map;

import org.apache.log4j.Logger;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Tuple;

public class RandmSentencebolt implements IRichBolt {

public void prepare(Map stormConf, TopologyContext context,
OutputCollector collector) {
// TODO Auto-generated method stub

}

public static Logger LOG = Logger.getLogger(RandmSentencebolt.class);

public void execute(Tuple input) {
// TODO Auto-generated method stub
System.out.println(input);
LOG.info(input);

}

public void cleanup() {
// TODO Auto-generated method stub

}

public void declareOutputFields(OutputFieldsDeclarer declarer) {
// TODO Auto-generated method stub
///declarer.declare(new Fields(""))

}

}

main function:
public class stormRun {
public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException, InterruptedException{
TopologyBuilder topbuilder = new TopologyBuilder();
topbuilder.setSpout(1,new RandomSentenceSpout(),5);
// topbuilder.setBolt(2, new bolthbase(), 5).fieldsGrouping(1, new Fields("word"));
topbuilder.setBolt(2, new RandmSentencebolt(), 5).fieldsGrouping(1, new Fields("word"));
Config conf = new Config();
conf.setDebug(true);
conf.setMaxTaskParallelism(10);
conf.setNumWorkers(10);
conf.setMaxSpoutPending(10);
StormSubmitter.submitTopology("stormtest", conf, topbuilder.createTopology());
Thread.sleep(10000);

}
}

i want to test spout to bolt !!!! bolt only print tuple!!!!

please help me to check code !!!

topbuilder.setSpout(1,new RandomSentenceSpout(),5);
// topbuilder.setBolt(2, new bolthbase(), 5).fieldsGrouping(1, new Fields("word"));
topbuilder.setBolt(2, new RandmSentencebolt(), 5).fieldsGrouping(1, new Fields("word"));
the code is right?