public static void main(String[] args) throws Exception {

        TopologyBuilder builder = new TopologyBuilder();

        builder.setSpout("spout", new RandomIntegerSpout());

        builder.setBolt("sumbolt", new WindowSumBolt().withWindow(new Count(5), new Count(3))

                .withMessageIdField("msgid"), 1).shuffleGrouping("spout");

        builder.setBolt("printer", new PrinterBolt(), 1).shuffleGrouping("sumbolt");

        Config conf = new Config();

        conf.setDebug(false);

        if (args != null && args.length > 0) {

            conf.setNumWorkers(1);

            StormSubmitter.submitTopologyWithProgressBar(args[0], conf, builder.createTopology());

        } else {

            LocalCluster cluster = new LocalCluster();

            StormTopology topology = builder.createTopology();

            cluster.submitTopology("test", conf, topology);

            Utils.sleep(40000);

            cluster.killTopology("test");

            cluster.shutdown();

        }

    }