class SentimentAnalysis extends AbstractTopology {

    @Override
    public StormTopology buildTopology() {
        TridentTopology topology = new TridentTopology();
        
        Stream rawTweets = null;

        switch (spoutType) {
            case 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, "tweet_str"));
                break;
            case SourceType.KAFKA:
                createKafkaSpout(spoutType, spoutType);
                break;
            default:
                throw new IllegalArgumentException("Must inform a spout type");
        }

        rawTweets.parallelismHint(spoutThreads)
                .shuffle();
                
        Stream tweets = rawTweets.each(new Fields("tweet_str"), new JsonParser(), new Fields("tweet"))
                .parallelismHint(parserThreads)
                .shuffle()
                .project(new Fields("tweet"));
                
        Stream parsedTweets = tweets.each(new Fields("tweet"), new TweetFilter(), new Fields("id", "text"))
                .parallelismHint(tweetFilterThreads)
                .shuffle()
                .project(new Fields("id", "text"));
                
        Stream cleanTweets = parsedTweets.each(new Fields("text"), new TextFilter(), new Fields("text_clean"))
                .parallelismHint(textFilterThreads)
                .shuffle()
                .project(new Fields("id", "text_clean"));
                
        Stream stemmedTweets = cleanTweets.each(new Fields("text_clean"), new Stemmer(), new Fields("text_stemmed"))
                .parallelismHint(stemmerThreads)
                .shuffle()
                .project(new Fields("id", "text_stemmed"));
                
        Stream posScores = stemmedTweets.each(new Fields("text_stemmed"), new SentimentTypeScorer(Type.Positive), new Fields("pos_score"))
                .parallelismHint(posScorerThreads)
                .shuffle()
                .project(new Fields("id", "text_stemmed", "pos_score"));
                
        Stream negScores = stemmedTweets.each(new Fields("text_stemmed"), new SentimentTypeScorer(Type.Negative), new Fields("neg_score"))
                .parallelismHint(negScorerThreads)
                .shuffle()
                .project(new Fields("id", "neg_score"));
        
        Stream joinedScores = topology.join(posScores, new Fields("id"), negScores, new Fields("id"), new Fields("id", "text", "pos_score", "neg_score"))
                .parallelismHint(sinkThreads);
                
        Stream scores = joinedScores.each(new Fields("pos_score", "neg_score"), new SentimentScorer(), new Fields("score"))
                .parallelismHint(scorerThreads)
                .shuffle();
                
        if (sinkType.equals(SinkType.CONSOLE)) {
            scores.each(scores.getOutputFields(), new Debug())
                    .parallelismHint(sinkThreads);
        }
        
        return topology.build();
    }

    public  SentimentAnalysis(String topologyName, Configuration config);

}