// Path: src/main/java/tutorial/storm/trident/testutil/TestUtils.java
// public class TestUtils {
//     public static TransactionalTridentKafkaSpout testTweetSpout(BrokerHosts hosts) {
//         TridentKafkaConfig kafkaConfig = new TridentKafkaConfig(hosts, "test", "storm");
//         kafkaConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
//         return new TransactionalTridentKafkaSpout(kafkaConfig);
//     }
// 
// }


import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.StormTopology;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.tuple.Fields;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import storm.kafka.*;
import storm.kafka.trident.TransactionalTridentKafkaSpout;
import storm.kafka.trident.TridentKafkaConfig;
import storm.trident.TridentTopology;
import tutorial.storm.trident.operations.*;
import tutorial.storm.trident.testutil.TestUtils;
import java.io.IOException;
import static com.google.common.base.Preconditions.checkArgument;
package tutorial.storm.trident;





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

        topology
                .newDRPCStream("ping");

        return topology.build();
    }


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


        if (args.length == 2) {
            // Ready & submit the topology
            String name = args[0];
            BrokerHosts hosts = new ZkHosts(args[1]);
// Path: src/main/java/tutorial/storm/trident/testutil/TestUtils.java
// public class TestUtils {
//     public static TransactionalTridentKafkaSpout testTweetSpout(BrokerHosts hosts) {
//         TridentKafkaConfig kafkaConfig = new TridentKafkaConfig(hosts, "test", "storm");
//         kafkaConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
//         return new TransactionalTridentKafkaSpout(kafkaConfig);
//     }
// 
// }


// Path: src/main/java/tutorial/storm/trident/Skeleton.java
import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.StormTopology;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.tuple.Fields;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import storm.kafka.*;
import storm.kafka.trident.TransactionalTridentKafkaSpout;
import storm.kafka.trident.TridentKafkaConfig;
import storm.trident.TridentTopology;
import tutorial.storm.trident.operations.*;
import tutorial.storm.trident.testutil.TestUtils;
import java.io.IOException;
import static com.google.common.base.Preconditions.checkArgument;

package tutorial.storm.trident;





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

        topology
                .newDRPCStream("ping");

        return topology.build();
    }


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


        if (args.length == 2) {
            // Ready & submit the topology
            String name = args[0];
            BrokerHosts hosts = new ZkHosts(args[1]);
            TransactionalTridentKafkaSpout kafkaSpout = TestUtils.testTweetSpout(hosts);