package tutorial.storm.trident;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.generated.StormTopology;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.tuple.Fields;
import com.google.common.collect.ImmutableList;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import storm.kafka.KafkaConfig;
import storm.kafka.StringScheme;
import storm.kafka.trident.TransactionalTridentKafkaSpout;
import storm.kafka.trident.TridentKafkaConfig;
import storm.trident.TridentTopology;
import storm.trident.operation.builtin.Count;
import storm.trident.operation.builtin.FirstN;
import storm.trident.operation.builtin.MapGet;
import storm.trident.operation.builtin.TupleCollectionGet;
import storm.trident.testing.FeederBatchSpout;
import storm.trident.testing.MemoryMapState;
import tutorial.storm.trident.operations.*;
import tutorial.storm.trident.testutil.SampleTweet;
import tutorial.storm.trident.testutil.TweetIngestor;

import java.io.File;
import java.io.IOException;
import java.util.UUID;

import static com.google.common.base.Preconditions.checkArgument;


/**
 * @author Enno Shioji (enno.shioji@peerindex.com)
 */
public class Skeleton {
    private static final Logger log = LoggerFactory.getLogger(Skeleton.class);
    public static StormTopology buildTopology(LocalDRPC drpc, TransactionalTridentKafkaSpout spout) throws IOException {
        TridentTopology topology = new TridentTopology();
        topology
                .newStream("tweets", spout)
                .each(new Fields("str"), new Print())
        ;


        topology
                .newDRPCStream("fake", drpc)
                .each(new Fields("args"), new Print())
        ;

        return topology.build();
    }


    public static void main(String[] args) throws Exception {
        Config conf = new Config();

        LocalDRPC drpc = new LocalDRPC();
        LocalCluster cluster = new LocalCluster();

        String testKafkaBrokerHost;
        if (args.length == 0) {
            final String tmpFile = "/tmp/"+UUID.randomUUID();
            new File(tmpFile).deleteOnExit();
            TweetIngestor ingestor = new TweetIngestor(tmpFile, "test", 12000);
            ingestor.startAndWait();
            testKafkaBrokerHost = "127.0.0.1:12000";
        }else{
            checkArgument(args.length == 1);
            testKafkaBrokerHost = args[0];
        }

        TransactionalTridentKafkaSpout tweetSpout = tweetSpout(testKafkaBrokerHost);
        cluster.submitTopology("hackaton", conf, buildTopology(drpc, tweetSpout));


        while (!Thread.currentThread().isInterrupted()) {
            Thread.sleep(3000);
//            System.out.println(drpc.execute("fake", "test"));
        }
    }

    private static TransactionalTridentKafkaSpout tweetSpout(String testKafkaBrokerHost) {
        KafkaConfig.BrokerHosts hosts = TridentKafkaConfig.StaticHosts.fromHostString(ImmutableList.of(testKafkaBrokerHost), 1);
        TridentKafkaConfig config = new TridentKafkaConfig(hosts, "test");
        config.scheme = new SchemeAsMultiScheme(new StringScheme());
        return new TransactionalTridentKafkaSpout(config);
    }
}

@@ -125,16 +125,54 @@
 alDRPC;%0A
+import backtype.storm.StormSubmitter;%0A
 import b
@@ -1404,24 +1404,8 @@
 ogy(
-LocalDRPC drpc, 
 Tran
@@ -1643,17 +1643,16 @@
      ;%0A%0A
-%0A
         
@@ -1696,84 +1696,14 @@
 am(%22
-fake%22, drpc)%0A                .each(new Fields(%22args%22), new Print())%0A        
+ping%22)
 ;%0A%0A 
@@ -1845,169 +1845,169 @@
 );%0A%0A
-        LocalDRPC drpc = new LocalDRPC();%0A        LocalCluster cluster = new LocalCluster();%0A%0A        String testKafkaBrokerHost;%0A        if (args.length == 0) %7B
+%0A        String testKafkaBrokerHost;%0A        if (args.length == 0) %7B%0A            // Executes the topology locally%0A%0A            // Ready the embedded Kafka broker
 %0A   
@@ -2285,122 +2285,56 @@
 0%22;%0A
- 
+%0A
        
-%7Delse%7B%0A            checkArgument(args.length == 1);%0A            testKafkaBrokerHost = args%5B0%5D;%0A        %7D%0A%0A
+     // Ready & submit the topology%0A    
     
@@ -2414,16 +2414,75 @@
 rHost);%0A
+            LocalCluster cluster = new LocalCluster();%0A    
         
@@ -2509,16 +2509,24 @@
 gy(%22
-hackaton
+trident-tutorial
 %22, c
@@ -2548,14 +2548,8 @@
 ogy(
-drpc, 
 twee
@@ -2555,25 +2555,28 @@
 etSpout));%0A%0A
-%0A
+    
         whil
@@ -2633,16 +2633,20 @@
         
+    
 Thread.s
@@ -2661,68 +2661,411 @@
 0);%0A
-//            System.out.println(drpc.execute(%22fake%22, %22test%22
+            %7D%0A%0A        %7Delse%7B%0A            // Submits the topology%0A            String topologyName = args%5B0%5D;%0A            testKafkaBrokerHost = args%5B1%5D;%0A            conf.setNumWorkers(8); // Our Vagrant environment has 8 workers%0A            TransactionalTridentKafkaSpout tweetSpout = tweetSpout(testKafkaBrokerHost);%0A            StormSubmitter.submitTopology(topologyName, conf, buildTopology(tweetSpout
 ));%0A
@@ -3074,16 +3074,19 @@
       %7D%0A
+%0A%0A%0A
     %7D%0A%0A 

Make it possible to submit the Skeleton topology to a cluster