[2/5] git commit: Add ExclamationTopologyWithMetrics to storm-starter

[pt...@apache.org]

Add ExclamationTopologyWithMetrics to storm-starter


Project: http://git-wip-us.apache.org/repos/asf/incubator-storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-storm/commit/0d6b438f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-storm/tree/0d6b438f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-storm/diff/0d6b438f

Branch: refs/heads/master
Commit: 0d6b438fff0ddeb254efe3d858ae7a29251ce289
Parents: d20ba2e
Author: troyding <di...@gmail.com>
Authored: Tue May 20 11:37:52 2014 +0800
Committer: troyding <di...@gmail.com>
Committed: Tue May 20 11:37:52 2014 +0800

----------------------------------------------------------------------
 .../starter/ExclamationTopologyWithMetrics.java | 88 ++++++++++++++++++++
 1 file changed, 88 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/0d6b438f/examples/storm-starter/src/jvm/storm/starter/ExclamationTopologyWithMetrics.java
----------------------------------------------------------------------
diff --git a/examples/storm-starter/src/jvm/storm/starter/ExclamationTopologyWithMetrics.java b/examples/storm-starter/src/jvm/storm/starter/ExclamationTopologyWithMetrics.java
new file mode 100644
index 0000000..96939c3
--- /dev/null
+++ b/examples/storm-starter/src/jvm/storm/starter/ExclamationTopologyWithMetrics.java
@@ -0,0 +1,88 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package storm.starter;
+
+import backtype.storm.Config;
+import backtype.storm.LocalCluster;
+import backtype.storm.StormSubmitter;
+import backtype.storm.metric.LoggingMetricsConsumer;
+import backtype.storm.task.OutputCollector;
+import backtype.storm.task.TopologyContext;
+import backtype.storm.testing.TestWordSpout;
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.topology.TopologyBuilder;
+import backtype.storm.topology.base.BaseRichBolt;
+import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.Tuple;
+import backtype.storm.tuple.Values;
+import backtype.storm.utils.Utils;
+
+import java.util.Map;
+
+/**
+ * This is a basic example of a Storm topology with metrics logging to the file.
+ */
+public class ExclamationTopologyWithMetrics {
+
+  public static class ExclamationBolt extends BaseRichBolt {
+    OutputCollector _collector;
+
+    @Override
+    public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
+      _collector = collector;
+    }
+
+    @Override
+    public void execute(Tuple tuple) {
+      _collector.emit(tuple, new Values(tuple.getString(0) + "!!!"));
+      _collector.ack(tuple);
+    }
+
+    @Override
+    public void declareOutputFields(OutputFieldsDeclarer declarer) {
+      declarer.declare(new Fields("word"));
+    }
+
+
+  }
+
+  public static void main(String[] args) throws Exception {
+    TopologyBuilder builder = new TopologyBuilder();
+
+    builder.setSpout("word", new TestWordSpout(), 10);
+    builder.setBolt("exclaim1", new ExclamationBolt(), 3).shuffleGrouping("word").setNumTasks(6);
+    builder.setBolt("exclaim2", new ExclamationBolt(), 2).shuffleGrouping("exclaim1").setNumTasks(6);
+
+    Config conf = new Config();
+    conf.setDebug(true);
+    conf.registerMetricsConsumer(LoggingMetricsConsumer.class);
+
+    if (args != null && args.length > 0) {
+      conf.setNumWorkers(3);
+      StormSubmitter.submitTopologyWithProgressBar(args[0], conf, builder.createTopology());
+    }
+    else {
+
+      LocalCluster cluster = new LocalCluster();
+      cluster.submitTopology("test", conf, builder.createTopology());
+      Utils.sleep(10000);
+      cluster.killTopology("test");
+      cluster.shutdown();
+    }
+  }
+}