ExclamationTopology.java

    import com.endgame.storm.metrics.statsd.StatsdMetricConsumer;

    ...

    TopologyBuilder builder = new TopologyBuilder();
    builder.setSpout("word", new TestWordSpout(), 10);
    builder.setBolt("exclaim1", new ExclamationBolt(), 3).shuffleGrouping("word");
    builder.setBolt("exclaim2", new ExclamationBolt(), 2).shuffleGrouping("exclaim1");
    
    //
    //  Configure the StatsdMetricConsumer
    //
    Map statsdConfig = new HashMap();
    statsdConfig.put(StatsdMetricConsumer.STATSD_HOST, "statsd.server.mydomain.com");
    statsdConfig.put(StatsdMetricConsumer.STATSD_PORT, 8125);
    statsdConfig.put(StatsdMetricConsumer.STATSD_PREFIX, "storm.metrics.");
    
    Config conf = new Config();
    conf.registerMetricsConsumer(StatsdMetricConsumer.class, statsdConfig, 2);
     
    if (args != null && args.length > 0) {
      conf.setNumWorkers(3);
      StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
    }
    else {
      LocalCluster cluster = new LocalCluster();
      cluster.submitTopology("test", conf, builder.createTopology());
      Utils.sleep(5*60*1000L);
      cluster.killTopology("test");
      cluster.shutdown();
    }