@@ -0,0 +1,2957 @@
+package tutorial.storm.trident;%0A%0Aimport backtype.storm.Config;%0Aimport backtype.storm.LocalCluster;%0Aimport backtype.storm.LocalDRPC;%0Aimport backtype.storm.generated.StormTopology;%0Aimport backtype.storm.spout.SchemeAsMultiScheme;%0Aimport backtype.storm.transactional.partitioned.IPartitionedTransactionalSpout;%0Aimport backtype.storm.tuple.Fields;%0Aimport backtype.storm.tuple.Values;%0Aimport com.google.common.base.Preconditions;%0Aimport com.google.common.collect.ImmutableList;%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0Aimport storm.kafka.KafkaConfig;%0Aimport storm.kafka.StringScheme;%0Aimport storm.kafka.trident.TransactionalTridentKafkaSpout;%0Aimport storm.kafka.trident.TridentKafkaConfig;%0Aimport storm.trident.TridentState;%0Aimport storm.trident.TridentTopology;%0Aimport 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.IOException;%0A%0A%0A/**%0A * @author Enno Shioji (enno.shioji@peerindex.com)%0A */%0Apublic class Skeleton %7B%0A    private static final Logger log = LoggerFactory.getLogger(Skeleton.class);%0A    public static StormTopology buildTopology(LocalDRPC drpc, TransactionalTridentKafkaSpout spout) throws IOException %7B%0A        TridentTopology topology = new TridentTopology();%0A        topology%0A                .newStream(%22tweets%22, spout)%0A                .each(new Fields(%22str%22), new Print())%0A        ;%0A%0A%0A        topology%0A                .newDRPCStream(%22fake%22, drpc)%0A                .each(new Fields(%22args%22), new Print())%0A        ;%0A%0A        return topology.build();%0A    %7D%0A%0A%0A    public static void main(String%5B%5D args) throws Exception %7B%0A        Config conf = new Config();%0A%0A        LocalDRPC drpc = new LocalDRPC();%0A        LocalCluster cluster = new LocalCluster();%0A%0A        String testKafkaBrokerHost = args%5B0%5D;%0A        TransactionalTridentKafkaSpout tweetSpout = tweetSpout(testKafkaBrokerHost);%0A        cluster.submitTopology(%22hackaton%22, conf, buildTopology(drpc, tweetSpout));%0A%0A%0A        while (!Thread.currentThread().isInterrupted()) %7B%0A            Thread.sleep(500);%0A//            System.out.println(drpc.execute(%22fake%22, %22test%22));%0A        %7D%0A    %7D%0A%0A    private static TransactionalTridentKafkaSpout tweetSpout(String testKafkaBrokerHost) %7B%0A//        TweetIngestor ingestor = new TweetIngestor(%22/tmp/kafka%22, %22test%22, 12000);%0A//        ingestor.startAndWait();%0A        KafkaConfig.BrokerHosts hosts = TridentKafkaConfig.StaticHosts.fromHostString(ImmutableList.of(testKafkaBrokerHost), 1);%0A        TridentKafkaConfig config = new TridentKafkaConfig(hosts, %22test%22);%0A        config.scheme = new SchemeAsMultiScheme(new StringScheme());%0A        return new TransactionalTridentKafkaSpout(config);%0A    %7D%0A%7D%0A

Add the skeleton