class TrendingTopics extends AbstractTopology {

    @Override
    public StormTopology buildTopology() {
        TridentTopology topology = new TridentTopology();
        Stream rawTweets = null;
        
        switch (spoutType) {
            case BaseConstants.SourceType.FILE:
                int maxBatchSize = config.getInt(Config.SPOUT_MAX_BATCH_SIZE, 100);
                String spoutPath = config.getString(Config.SPOUT_PATH);
                rawTweets = topology.newStream("spout", new FileSpout(maxBatchSize, spoutPath));
                break;
            case BaseConstants.SourceType.KAFKA:
                createKafkaSpout(spoutType, spoutType);
                break;
            default:
                throw new IllegalArgumentException("Must inform a spout type");
        }
        
        rawTweets = rawTweets.parallelismHint(spoutThreads)
                .shuffle();
        
        Stream tweets = rawTweets.each(new Fields("tweet_str"), new JsonParser(), new Fields("tweet"))
                .parallelismHint(parserThreads)
                .shuffle()
                .project(new Fields("tweet"));
        
        GroupedStream topics = tweets.each(new Fields("tweet"), new TopicExtractor(), new Fields("topic"))
                .parallelismHint(topicExtractorThreads)
                .groupBy(new Fields("topic"));
        
        Stream counts = topics.persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count"))
                .parallelismHint(counterThreads)
                .newValuesStream();
        
        Stream trending = counts.persistentAggregate(new MemoryMapState.Factory(), new Fields("topic", "count"), new MaxCount(), new Fields("top"))
                .newValuesStream()
                .parallelismHint(trendingThreads);
        
        //Stream trending = counts.batchGlobal().partitionAggregate(new Fields("topic", "count"), new MaxCount2(), new Fields("topic", "count"));
        
        if (sinkType.equals(SinkType.CONSOLE)) {
            trending.each(new Fields("top"), new Debug())
                    .parallelismHint(sinkThreads);
        }
        
        return topology.build();
    }

    public  TrendingTopics(String topologyName, Configuration config);

}