<\/span><\/h2>\nA spout in its simplest form is listed below. We initialize the collector object in the open() method, which is used for emitting randomnly-generated, base64-encoded strings. We use a simple mechanism for limiting the number of tuples that can be active (i.e. not yet acked) in the topology: this is to avoid filling the internal queues to the point of overflow (which can lead to out-of-memory exceptions). We can circumnavigate this simple queue machanism by setting maxPendingMsgs to UNLIMITED_PENDING in the constructor. There is no replay (the class is intentionally simplified), as failed tuples are simply removed from the pending queue. If a queue limit has been specified (i.e. maxPendingMsgs <> UNLIMITED_PENDING), and this limit has been reached, then nothing new is emitted from the spout until there is space in the queue.<\/p>\n
public class SimpleSpout extends BaseRichSpout {\r\n\r\n\tprivate final List<Object> pending;\r\n\r\n\tprivate SpoutOutputCollector collector;\r\n\r\n\tprivate final int maxPendingMsgs;\r\n\r\n\tprivate final static Random rand = new Random();\r\n\r\n\tpublic final static int UNLIMITED_PENDING = -1;\r\n\r\n\tpublic SimpleSpout(int maxPendingMsgs) {\r\n\r\n\t\tthis.pending = new LinkedList<Object>();\r\n\r\n\t\tthis.maxPendingMsgs = maxPendingMsgs;\r\n\r\n\t}\r\n\r\n\t@Override\r\n\r\n\tpublic void open(@SuppressWarnings(\"rawtypes\") Map conf, TopologyContext context, SpoutOutputCollector collector) {\r\n\r\n\t\tthis.collector = collector;\r\n\r\n\t}\r\n\r\n\t@Override\r\n\r\n\tpublic void nextTuple() {\r\n\r\n\t\tif (UNLIMITED_PENDING == maxPendingMsgs || pending.size() < maxPendingMsgs) {\r\n\r\n\t\t\tString s = String.valueOf(rand.nextLong());\r\n\r\n\t\t\tString b64 = Base64.encodeBase64String(s.getBytes(Charsets.UTF_8));\r\n\r\n\t\t\tif (UNLIMITED_PENDING == maxPendingMsgs) {\r\n\r\n\t\t\t\t\/* turn off acking if we are not worried about queue overflow *\/\r\n\r\n\t\t\t\tcollector.emit(new Values(b64));\r\n\r\n\t\t\t} else {\r\n\r\n\t\t\t\tUUID uuid = UUID.randomUUID();\r\n\r\n\t\t\t\tcollector.emit(new Values(b64), uuid);\r\n\r\n\t\t\t\tpending.add(uuid);\r\n\r\n\t\t\t}\r\n\r\n\t\t}\r\n\r\n\t}\r\n\r\n\t@Override\r\n\r\n\tpublic void declareOutputFields(OutputFieldsDeclarer declarer) {\r\n\r\n\t\tdeclarer.declare(new Fields(\"emitted\"));\r\n\r\n\t}\r\n\r\n\t@Override\r\n\r\n\tpublic void ack(Object uuid) {\r\n\r\n\t\tsuper.ack(uuid);\r\n\r\n\t\tpending.remove(uuid);\r\n\r\n\t}\r\n\r\n\t@Override\r\n\r\n\tpublic void fail(Object uuid) {\r\n\r\n\t\tsuper.fail(uuid);\r\n\r\n\t\tpending.remove(uuid);\r\n\r\n\t}\r\n\r\n}<\/pre>\nA simple bolt that consumes data from this spout is listed below.<\/p>\n
public class ReadMapBolt extends BaseBasicBolt {\r\n\r\n @Override\r\n\r\n public void execute(Tuple tuple, BasicOutputCollector collector) {\r\n\r\n long start = System.nanoTime();\r\n\r\n String b64 = tuple.getValues().get(0).toString();\r\n\r\n byte[] bb = Base64.decodeBase64(b64);\r\n\r\n collector.emit(new Values(\"ok\"));\r\n\r\n }\r\n\r\n @Override\r\n\r\n public void declareOutputFields(OutputFieldsDeclarer ofd) {\r\n\r\n ofd.declare(new Fields(\"parseoutput\"));\r\n\r\n }\r\n\r\n}<\/pre>\nNote that the information listed in the declareOutputFields() methods must be consistent across the topology (i.e. either explicitly when retrieving tuple fields by name, or implicitly when doing so by position, as above) otherwise the topology will throw an exception on deployment.<\/p>\n
Lastly, our topology links the spout and bolt together:<\/p>\n
public class SimpleWriterAndReader {\r\n\r\n public static void main(String[] args) throws Exception {\r\n\r\n TopologyBuilder builder = new TopologyBuilder();\r\n\r\n builder.setSpout(\"spout\", new SimpleSpout(64), 1);\r\n\r\n builder.setBolt(\"bolt\", new ReadMapBolt(), 4).shuffleGrouping(\"spout\").setNumTasks(4);\r\n\r\n Config conf = new Config();\r\n\r\n conf.setDebug(false);\r\n\r\n conf.setNumWorkers(1);\r\n\r\n LocalCluster cluster = new LocalCluster();\r\n\r\n cluster.submitTopology(\"test_topology\", conf, builder.createTopology());\r\n\r\n Thread.sleep(1000 * 60 * 15);\r\n\r\n cluster.shutdown();\r\n\r\n }\r\n\r\n}<\/pre>\nThis topology launches in local mode and shuts down after 15 minutes.<\/p>\n
We now move on to considering some not-so-trivial aspects of a storm topology that may be of interest.<\/p>\n