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

        TopologyBuilder builder = new TopologyBuilder();

        BaseWindowedBolt bolt = new SlidingWindowSumBolt()

                .withWindow(new Duration(5, TimeUnit.SECONDS), new Duration(3, TimeUnit.SECONDS))

                .withTimestampField("ts")

                .withLag(new Duration(5, TimeUnit.SECONDS));

        builder.setSpout("integer", new RandomIntegerSpout(), 1);

        builder.setBolt("slidingsum", bolt, 1).shuffleGrouping("integer");

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

        Config conf = new Config();

        conf.setDebug(true);



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

            conf.setNumWorkers(1);

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

        } else {

            LocalCluster cluster = new LocalCluster();

            cluster.submitTopology("test", conf, builder.createTopology());

            Utils.sleep(40000);

            cluster.killTopology("test");

            cluster.shutdown();

        }

    }