Stream closed error - Storm

Question

I am trying to read a line in a file, in the spout, and then send it to the bolts but i keep getting a stream closed error. Do I close it wrong or what is the problem here?

public class InputSpout extends BaseRichSpout {
    private SpoutOutputCollector collector;
    public void
    declareOutputFields( OutputFieldsDeclarer declarer) {
        declarer.declare( new Fields("logfile"));
    }
    private FileReader fileReader;
    private boolean completed = false;
    private TopologyContext context;
    @Override
    public void open( Map config, TopologyContext context, SpoutOutputCollector collector) { 
        try {
            this.context = context;
            this.fileReader = new FileReader(("logfile.txt").toString());
        } catch (FileNotFoundException e) {
            throw new RuntimeException("Error reading file "
                    + ("logfile"));
        }
        this.collector = collector;
    }
    public void nextTuple() {


        if (completed) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {

            }
        }
        String str;
        BufferedReader reader = new BufferedReader(fileReader);
        str = null;
        try {
            str = reader.readLine();
        } catch (IOException e1) {
            e1.printStackTrace();
        }
        try {
            while (str != null) {
                this.collector.emit(new Values(str));
                str = reader.readLine();
            }

        } catch (Exception e) {
            throw new RuntimeException("Error reading typle", e);
        } finally {
            completed = true;
        }
        try {
            reader.close();
            fileReader.close();
        } catch (IOException e) {
            e.printStackTrace();
        }

This is the error I am getting:

java.io.IOException: Stream closed at sun.nio.cs.StreamDecoder.ensureOpen(StreamDecoder.java:46) at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:147) at java.io.InputStreamReader.read(InputStreamReader.java:184) at java.io.BufferedReader.fill(BufferedReader.java:154) at java.io.BufferedReader.readLine(BufferedReader.java:317) at java.io.BufferedReader.readLine(BufferedReader.java:382) at myStorm.InputSpout.nextTuple(InputSpout.java:52) at backtype.storm.daemon.executor$fn__4654$fn__4669$fn__4698.invoke(executor.clj:565) at backtype.storm.util$async_loop$fn__458.invoke(util.clj:463) at clojure.lang.AFn.run(AFn.java:24) at java.lang.Thread.run(Thread.java:745)

Answer 1

You're closing the InputStream used by fileReader in nextTuple making it unavailable for subsequent calls. There's no need to close this Reader - just close a single BufferedReader when all data has been read.