// Path: src/main/java/tutorial/storm/trident/operations/ExtractLocation.java
// public class ExtractLocation extends BaseFunction {
// 
//     @Override
//     public void execute(TridentTuple tuple, TridentCollector collector) {
//         Status status = (Status) tuple.get(0);
//         Content content = (Content) tuple.get(1);
// 
//         collector.emit(new Values(status.getPlace().getCountryCode(), content.getContentName()));
//     }
// }
// 
// Path: src/main/java/tutorial/storm/trident/operations/OnlyGeo.java
// public class OnlyGeo extends BaseFilter {
//     @Override
//     public boolean isKeep(TridentTuple tuple) {
//         Status status = (Status) tuple.get(0);
//         return !(null == status.getPlace() || null == status.getPlace().getCountryCode());
//     }
// }
// 
// Path: src/main/java/tutorial/storm/trident/operations/OnlyHashtags.java
// public class OnlyHashtags extends BaseFilter {
//     @Override
//     public boolean isKeep(TridentTuple tuple) {
//         Content content = (Content)tuple.get(0);
//         return "hashtag".equals(content.getContentType());
//     }
// }
// 
// Path: src/main/java/tutorial/storm/trident/operations/ParseTweet.java
// public class ParseTweet extends BaseFunction {
//     private static final Logger log = LoggerFactory.getLogger(ParseTweet.class);
// 
//     private ContentExtracter extracter;
// 
//     @Override
//     public void execute(TridentTuple tuple, TridentCollector collector) {
//         if(extracter == null) extracter = new ContentExtracter();
// 
//         String rawTweetJson = (String)tuple.get(0);
//         Status parsed = parse(rawTweetJson);
//         User user = parsed.getUser();
// 
//         for (Content content : extracter.extract(parsed)) {
//             collector.emit(new Values(parsed, content, user));
//         }
//     }
// 
//     private Status parse(String rawJson){
//         try {
//             Status parsed = DataObjectFactory.createStatus(rawJson);
//             return parsed;
//         } catch (TwitterException e) {
//             log.warn("Invalid tweet json -> " + rawJson, e);
//             return null;
//         }
//     }
// }
// 
// 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.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.StormTopology;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.tuple.Fields;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import storm.kafka.BrokerHosts;
import storm.kafka.KafkaConfig;
import storm.kafka.StringScheme;
import storm.kafka.ZkHosts;
import storm.kafka.trident.TransactionalTridentKafkaSpout;
import storm.kafka.trident.TridentKafkaConfig;
import storm.trident.TridentState;
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.MemoryMapState;
import tutorial.storm.trident.operations.ExtractLocation;
import tutorial.storm.trident.operations.OnlyGeo;
import tutorial.storm.trident.operations.OnlyHashtags;
import tutorial.storm.trident.operations.ParseTweet;
import tutorial.storm.trident.testutil.TestUtils;
import java.io.IOException;
package tutorial.storm.trident.example;



/**
*
*/
public class TopHashtagByCountry {

    public static StormTopology buildTopology(TransactionalTridentKafkaSpout spout) throws IOException {

        TridentTopology topology = new TridentTopology();
        TridentState count =
        topology
                .newStream("tweets", spout)
                .each(new Fields("str"), new ParseTweet(), new Fields("status", "content", "user"))
                .project(new Fields("content", "user", "status"))
                .each(new Fields("content"), new OnlyHashtags())
                .each(new Fields("status"), new OnlyGeo())
// Path: src/main/java/tutorial/storm/trident/operations/ExtractLocation.java
// public class ExtractLocation extends BaseFunction {
// 
//     @Override
//     public void execute(TridentTuple tuple, TridentCollector collector) {
//         Status status = (Status) tuple.get(0);
//         Content content = (Content) tuple.get(1);
// 
//         collector.emit(new Values(status.getPlace().getCountryCode(), content.getContentName()));
//     }
// }
// 
// Path: src/main/java/tutorial/storm/trident/operations/OnlyGeo.java
// public class OnlyGeo extends BaseFilter {
//     @Override
//     public boolean isKeep(TridentTuple tuple) {
//         Status status = (Status) tuple.get(0);
//         return !(null == status.getPlace() || null == status.getPlace().getCountryCode());
//     }
// }
// 
// Path: src/main/java/tutorial/storm/trident/operations/OnlyHashtags.java
// public class OnlyHashtags extends BaseFilter {
//     @Override
//     public boolean isKeep(TridentTuple tuple) {
//         Content content = (Content)tuple.get(0);
//         return "hashtag".equals(content.getContentType());
//     }
// }
// 
// Path: src/main/java/tutorial/storm/trident/operations/ParseTweet.java
// public class ParseTweet extends BaseFunction {
//     private static final Logger log = LoggerFactory.getLogger(ParseTweet.class);
// 
//     private ContentExtracter extracter;
// 
//     @Override
//     public void execute(TridentTuple tuple, TridentCollector collector) {
//         if(extracter == null) extracter = new ContentExtracter();
// 
//         String rawTweetJson = (String)tuple.get(0);
//         Status parsed = parse(rawTweetJson);
//         User user = parsed.getUser();
// 
//         for (Content content : extracter.extract(parsed)) {
//             collector.emit(new Values(parsed, content, user));
//         }
//     }
// 
//     private Status parse(String rawJson){
//         try {
//             Status parsed = DataObjectFactory.createStatus(rawJson);
//             return parsed;
//         } catch (TwitterException e) {
//             log.warn("Invalid tweet json -> " + rawJson, e);
//             return null;
//         }
//     }
// }
// 
// 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/example/TopHashtagByCountry.java
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.StormTopology;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.tuple.Fields;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import storm.kafka.BrokerHosts;
import storm.kafka.KafkaConfig;
import storm.kafka.StringScheme;
import storm.kafka.ZkHosts;
import storm.kafka.trident.TransactionalTridentKafkaSpout;
import storm.kafka.trident.TridentKafkaConfig;
import storm.trident.TridentState;
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.MemoryMapState;
import tutorial.storm.trident.operations.ExtractLocation;
import tutorial.storm.trident.operations.OnlyGeo;
import tutorial.storm.trident.operations.OnlyHashtags;
import tutorial.storm.trident.operations.ParseTweet;
import tutorial.storm.trident.testutil.TestUtils;
import java.io.IOException;

package tutorial.storm.trident.example;



/**
*
*/
public class TopHashtagByCountry {

    public static StormTopology buildTopology(TransactionalTridentKafkaSpout spout) throws IOException {

        TridentTopology topology = new TridentTopology();
        TridentState count =
        topology
                .newStream("tweets", spout)
                .each(new Fields("str"), new ParseTweet(), new Fields("status", "content", "user"))
                .project(new Fields("content", "user", "status"))
                .each(new Fields("content"), new OnlyHashtags())
                .each(new Fields("status"), new OnlyGeo())
                .each(new Fields("status", "content"), new ExtractLocation(), new Fields("country", "contentName"))