/**
 * This topology used for watching SegmentFault's haproxy log.
 */
package com.segmentfault.storm;

import storm.kafka.*;

import java.util.ArrayList;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Tuple;

/**
 * @author tairy
 * @date 2016-03-13
 * @version 0.0.1
 */
public class SampleTopology 
{
	public static class PrinterBolt extends BaseBasicBolt {
		/**
		 * serialVersionUID
		 */
		private static final long serialVersionUID = 3179969158010282366L;

		public void declareOutputFields(OutputFieldsDeclarer declarer) {
        }

        public void execute(Tuple tuple, BasicOutputCollector collector) {
            System.out.println(tuple.toString());
        }
    }
	
	public static void main(String args[]) throws Exception
	{
		String zookeeperHost = "docker.tairy.me:2181";
		ZkHosts zkHosts = new ZkHosts(zookeeperHost);
		SpoutConfig spoutConfig = new SpoutConfig(zkHosts, "sf-log", "/kafkastorm", "id");
		spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
		spoutConfig.zkServers = new ArrayList<String>() {{  
            add("docker.tairy.me");  
        }};  
        spoutConfig.zkPort = 2181;
		KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);
		
		TopologyBuilder builder = new TopologyBuilder();
		builder.setSpout("sf-log", kafkaSpout,  1);
		builder.setBolt("print", new PrinterBolt()).shuffleGrouping("sf-log");
		LocalCluster cluster = new LocalCluster();
		Config config = new Config();
		cluster.submitTopology("kafka-test", config, builder.createTopology());
		
		Thread.sleep(6000);
	}
}

@@ -599,17 +599,20 @@
 logy %0A%7B%0A
-%09
+    
 public s
@@ -659,18 +659,24 @@
 cBolt %7B%0A
-%09%09
+        
 /**%0A%09%09 *
@@ -699,18 +699,24 @@
 D%0A%09%09 */%0A
-%09%09
+        
 private 
@@ -775,18 +775,24 @@
 2366L;%0A%0A
-%09%09
+        
 public v
@@ -1000,19 +1000,21 @@
 %7D%0A    %7D%0A
-%09%0A%09
+%0A    
 public s
@@ -1065,13 +1065,22 @@
 ion%0A
-%09%7B%0A%09%09
+    %7B%0A        
 Stri
@@ -1122,18 +1122,24 @@
 :2181%22;%0A
-%09%09
+        
 ZkHosts 
@@ -1176,18 +1176,24 @@
 rHost);%0A
-%09%09
+        
 SpoutCon
@@ -1267,18 +1267,24 @@
  %22id%22);%0A
-%09%09
+        
 spoutCon
@@ -1341,18 +1341,24 @@
 eme());%0A
-%09%09
+        
 spoutCon
@@ -1489,18 +1489,24 @@
 = 2181;%0A
-%09%09
+        
 KafkaSpo
@@ -1550,21 +1550,25 @@
 onfig);%0A
-%09%09%0A%09%09
+%0A        
 Topology
@@ -1608,18 +1608,24 @@
 lder();%0A
-%09%09
+        
 builder.
@@ -1660,18 +1660,24 @@
 t,  1);%0A
-%09%09
+        
 builder.
@@ -1739,18 +1739,24 @@
 -log%22);%0A
-%09%09
+        
 LocalClu
@@ -1790,18 +1790,24 @@
 ster();%0A
-%09%09
+        
 Config c
@@ -1828,18 +1828,24 @@
 nfig();%0A
-%09%09
+        
 cluster.
@@ -1912,13 +1912,17 @@
 ));%0A
-%09%09%0A%09%09
+%0A        
 Thre
@@ -1937,13 +1937,16 @@
 (6000);%0A
-%09
+    
 %7D%0A%7D%0A

change format!