Apache Storm - implementation for a spout that connects through a TCP socket to a server

SocketSpout.java

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

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.Socket;
import java.util.Map;

/**
* Implementation for a Spout that connects to a server
* using a TCP Socket.
*/
public class SocketSpout extends BaseRichSpout {
    private SpoutOutputCollector collector;
    private String host;
    private int port;

    private BufferedReader in;

    public SocketSpout(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public void declareOutputFields(OutputFieldsDeclarer decl) {
        // declare your output fields
        // decl.declare(new Fields(...));
    }

    public void open(Map map, TopologyContext ctx, SpoutOutputCollector collector) {
        this.collector = collector;
        try {
            Socket s = new Socket(host, port);
            in = new BufferedReader(
                    new InputStreamReader(s.getInputStream()));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void nextTuple() {
        try {
            String raw = in.readLine();

            if (raw == null) {
                // make the topology die
                throw new SocketSpoutKO();
            }

            // parse the raw string; e.g.
            // raw = raw.replace(...);
            // String[] fields = raw.split(",");

            // emit using collector
            // collector.emit(new Values(...));
            
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private class SocketSpoutKO extends RuntimeException { }
}