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 tutorial.storm.trident.operations.Print;

import java.io.IOException;


/**
 * @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 = args[0];
        TransactionalTridentKafkaSpout tweetSpout = tweetSpout(testKafkaBrokerHost);
        cluster.submitTopology("hackaton", conf, buildTopology(drpc, tweetSpout));


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

    private static TransactionalTridentKafkaSpout tweetSpout(String testKafkaBrokerHost) {
        // You can start your own Kafka broker that emits tweets. See {@link TweetIngestor} for details
//        TweetIngestor ingestor = new TweetIngestor("/tmp/kafka", "test", 12000);
//        ingestor.startAndWait();
        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);
    }
}

@@ -582,76 +582,554 @@
 ort 
-tutorial.storm.trident.operations.Print;%0A%0Aimport java.io.IOException
+storm.trident.operation.builtin.Count;%0Aimport storm.trident.operation.builtin.FirstN;%0Aimport storm.trident.operation.builtin.MapGet;%0Aimport storm.trident.operation.builtin.TupleCollectionGet;%0Aimport storm.trident.testing.FeederBatchSpout;%0Aimport storm.trident.testing.MemoryMapState;%0Aimport tutorial.storm.trident.operations.*;%0Aimport tutorial.storm.trident.testutil.SampleTweet;%0A%0Aimport java.io.File;%0Aimport java.io.IOException;%0Aimport java.nio.file.Files;%0Aimport java.util.UUID;%0A%0Aimport static com.google.common.base.Preconditions.checkArgument
 ;%0A%0A%0A
@@ -1996,19 +1996,433 @@
 Host
- = args%5B0%5D;
+;%0A        if (args.length == 0) %7B%0A            final String tmpFile = %22/tmp/%22+UUID.randomUUID();%0A            new File(tmpFile).deleteOnExit();%0A            TweetIngestor ingestor = new TweetIngestor(tmpFile, %22test%22, 12000);%0A            ingestor.startAndWait();%0A            testKafkaBrokerHost = %22127.0.0.1:12000%22;%0A        %7Delse%7B%0A            checkArgument(args.length == 1);%0A            testKafkaBrokerHost = args%5B0%5D;%0A        %7D%0A
 %0A   
@@ -2675,16 +2675,15 @@
 eep(
-5
+30
 00);%0A
-//
     
@@ -2852,230 +2852,8 @@
 ) %7B%0A
-        // You can start your own Kafka broker that emits tweets. See %7B@link TweetIngestor%7D for details%0A//        TweetIngestor ingestor = new TweetIngestor(%22/tmp/kafka%22, %22test%22, 12000);%0A//        ingestor.startAndWait();%0A
     

Add code to start kafka broker locally