class WordCount extends AbstractTopology {

    @Override
    public StormTopology buildTopology() {
        TridentTopology topology = new TridentTopology();
        
        Stream sentences = null;
        
        switch (spoutType) {
            case BaseConstants.SourceType.FILE:
                int maxBatchSize = config.getInt(Config.SPOUT_MAX_BATCH_SIZE, 100);
                String spoutPath = config.getString(Config.SPOUT_PATH);
                sentences = 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");
        }
        
        sentences = sentences.parallelismHint(spoutThreads);
                
        GroupedStream words = sentences.each(new Fields("sentence"), new Split(), new Fields("word"))
                .parallelismHint(splitterThreads)
                .groupBy(new Fields("word"));
                
        Stream counts = words.persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count"))
                .parallelismHint(counterThreads)
                .newValuesStream();
        
        if (sinkType.equals(SinkType.CONSOLE)) {
            counts.each(new Fields("word", "count"), new Debug())
                    .parallelismHint(sinkThreads);
        }

        return topology.build();
    }

    public  WordCount(String topologyName, Configuration config);

}