OrderRegionalTopology.java

public class OrderRegionalTopology {
 
    public static void main(String[] args) throws Exception {
        TopologyBuilder builder = new TopologyBuilder();
 
        builder.setSpout("orderSource", new OrderEnabledMockSpout(), 5);
        builder.setBolt("regional", new RegionalStatisticsBolt(), 20)
                .fieldsGrouping("orderSource", new Fields("IP"));
 
        Config conf = new Config();
        conf.setDebug(true);
 
        if(args!=null && args.length > 0) {
            conf.setNumWorkers(3);
            StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
        } else {
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("regional", conf, builder.createTopology());
            //Utils.sleep(10000);
            //cluster.killTopology("regional");
            //cluster.shutdown();
        }
    }
}