public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException {
		
		if(args.length < 1 || !ALLOWED_OPTION.containsKey(args[0])){
			System.out.println("Usage \"TwitterWordCountTopology CountingMode\"");
			System.out.println("Where CountingMode is: ");
			for(String key: ALLOWED_OPTION.keySet()){
				System.out.println(key + ", " + ALLOWED_OPTION.get(key));
			}
			System.exit(-1);
		}
		
		TopologyBuilder builder = new TopologyBuilder();
		
		//use TwitterSampleSpout
		String twitterSampleSpoutId = "sample_tweet";
		int twitterSampleSpoutParallelism = 1;
		builder.setSpout(twitterSampleSpoutId, new TwitterSampleSpout(), twitterSampleSpoutParallelism);
		
		//use SplitTweet Bolt
		String splitTweetBoltId = "split_tweet";
		int splitTweetParallelism = 10;
		builder.setBolt(splitTweetBoltId, new SplitTweet(), splitTweetParallelism)
				.shuffleGrouping(twitterSampleSpoutId);
			
		//use CountWord Bolt
		//better code: use builder class to build specific bolt
		BaseRichBolt countWordBolt = null;
		String countWordBoltId = null;
		int numArgument = 0;
		if(args[0].equals(NORMAL_MODE)){
			countWordBolt = new CountWord();
			countWordBoltId = "count_word";
			numArgument = 1;
		}else if (args[0].equals(FED_MODE)){
			int tickFrequency = 0;
			int decayAlpha = 0; 
			numArgument = 3;
			
			if(args.length < 3){
				System.out.println(FED_MODE_ERROR_MSG);
				System.exit(-1);
			}
			
			try {
				decayAlpha = Integer.parseInt(args[1]);
				tickFrequency = Integer.parseInt(args[2]);
			} catch (NumberFormatException e) {
				System.out.println("Error in parsing decay_alpha or output_frequency");
				System.out.println(FED_MODE_ERROR_MSG);
				System.exit(-1);
			}

			countWordBolt = new ForwardDecayCountWord(tickFrequency, decayAlpha);
			countWordBoltId = "fwd_decay_count_word";
		}else{
			System.out.println(args[0] + " is a valid option but not yet implemented");
			return;
		}

		int countWordBoltParallelism = 10;
		builder.setBolt(countWordBoltId, countWordBolt, countWordBoltParallelism)
				.fieldsGrouping(splitTweetBoltId, new Fields("word"));
		
		//use PrinterCount Bolt to print the result
		String printerBoltId = "printer_bolt";
		int printerBoltPar = 10;
		builder.setBolt(printerBoltId, new PrintCount(), printerBoltPar)
			.fieldsGrouping(countWordBoltId, new Fields("word"));
		
		Config conf = new Config();
		conf.setDebug(false);
		
		if(args != null && args.length > numArgument){
			conf.setNumWorkers(3);
			
			StormSubmitter.submitTopology(args[numArgument], conf, builder.createTopology());
		}else{
			conf.setMaxTaskParallelism(3);
			
			String topologyName = "word_count_topology";
			LocalCluster cluster = new LocalCluster();
			cluster.submitTopology(topologyName , conf, builder.createTopology());
			
			Utils.sleep(90*1000);
			
			cluster.killTopology(topologyName);
			cluster.shutdown();			
		}
	}