//package org.apache.storm.starter;

import java.util.Map;

import java.io.File;
import java.util.Scanner;
import java.util.List;
import java.util.ArrayList;
import java.io.FileNotFoundException;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichSpout;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

public class WordReaderSpout implements IRichSpout {
    private SpoutOutputCollector collector;
    boolean isFinished = false;

    public List<String> readData(String filename){
      List<String> s = new ArrayList<String>();
      String token1 = "";
      try {
         Scanner infile = new Scanner(new File(filename)).useDelimiter("\n");
         while (infile.hasNext()) {
	      token1 = infile.next();
	      s.add(token1);
         }
         infile.close();
      } catch (FileNotFoundException e) {
         e.printStackTrace();
      }
      return s;
    }


    @Override
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void close(){
    }

    @Override

    public void activate() {

        // TODO Auto-generated method stub

    }

    @Override

    public void deactivate() {

        // TODO Auto-generated method stub

    }
    @Override

    public void nextTuple() {
    List<String> sentences = readData("../data.txt");
        if (!isFinished) {
            for (String sentence: sentences) {
                for (String word: sentence.split(" ")) {
                    this.collector.emit(new Values(word));
                }
            }
            isFinished = true;
        } else {
            this.close();
        }
    }

    @Override

    public void ack(Object msgId) {

        // TODO Auto-generated method stub

    }

    @Override

    public void fail(Object msgId) {

        // TODO Auto-generated method stub

    }

    @Override

    public void declareOutputFields(OutputFieldsDeclarer declarer) {

        declarer.declare(new Fields("word"));

    }

    @Override

    public Map < String, Object > getComponentConfiguration() {

        // TODO Auto-generated method stub

        return null;

    }
}