// Path: src/main/java/net/pixelcop/sewer/StatusProvider.java
// public interface StatusProvider {
// 
//   /**
//    * Channel is not open (default status)
//    */
//   public static final int CLOSED    = 0;
// 
//   /**
//    * Channel is in the process of closing down
//    */
//   public static final int CLOSING   = 1;
// 
//   /**
//    * Channel is in the process of opening
//    */
//   public static final int OPENING   = 2;
// 
//   /**
//    * Channel is active and Events are flowing through it
//    */
//   public static final int FLOWING   = 3;
// 
//   /**
//    * Channel is in an error state and Events are not flowing through it. This state could be set
//    * during any operation: open, close, append, etc.
//    */
//   public static final int ERROR     = -1;
// 
//   public int getStatus();
// 
// }
// 
// Path: src/main/java/net/pixelcop/sewer/Sink.java
// public abstract class Sink extends Plumbing {
// 
//   private static final Logger LOG = LoggerFactory.getLogger(Sink.class);
// 
//   protected Sink subSink = null;
// 
//   /**
//    * Append an Event to the Sink
//    * @param event
//    * @throws IOException
//    */
//   public abstract void append(Event event) throws IOException;
// 
//   public void setSubSink(Sink subSink) {
//     this.subSink = subSink;
//   }
// 
//   public Sink getSubSink() {
//     return subSink;
//   }
// 
//   /**
//    * Create the SubSink, if possible
//    *
//    * @return True if a SubSink was created, false otherwise.
//    */
//   public boolean createSubSink() {
//     if (getSinkFactory() == null) {
//       return false;
//     }
// 
//     subSink = getSinkFactory().build();
//     return true;
//   }
// 
//   @Override
//   protected void finalize() throws Throwable {
//     if (getStatus() != CLOSED) {
//       try {
//         close();
//       } catch (Throwable t) {
//         LOG.warn("Caught during finalizer close(): " + t.getMessage(), t);
//         // rethrow? this should be good enough since the object was disposed of anyway
//       }
//     }
//   }
// 
// }
// 
// Path: src/main/java/net/pixelcop/sewer/StatusProvider.java
// public interface StatusProvider {
// 
//   /**
//    * Channel is not open (default status)
//    */
//   public static final int CLOSED    = 0;
// 
//   /**
//    * Channel is in the process of closing down
//    */
//   public static final int CLOSING   = 1;
// 
//   /**
//    * Channel is in the process of opening
//    */
//   public static final int OPENING   = 2;
// 
//   /**
//    * Channel is active and Events are flowing through it
//    */
//   public static final int FLOWING   = 3;
// 
//   /**
//    * Channel is in an error state and Events are not flowing through it. This state could be set
//    * during any operation: open, close, append, etc.
//    */
//   public static final int ERROR     = -1;
// 
//   public int getStatus();
// 
// }


import static net.pixelcop.sewer.StatusProvider.*;
import java.io.BufferedInputStream;
import java.io.DataInputStream;
import java.io.EOFException;
import java.io.IOException;
import java.net.Socket;
import net.pixelcop.sewer.Sink;
import net.pixelcop.sewer.StatusProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
package net.pixelcop.sewer.source;





public abstract class TCPReaderThread extends Thread {

  private static final Logger LOG = LoggerFactory.getLogger(TCPReaderThread.class);

  protected final Socket socket;
  protected final Sink sink;
  protected DataInputStream in;
  protected final StatusProvider statusProvider;

// Path: src/main/java/net/pixelcop/sewer/StatusProvider.java
// public interface StatusProvider {
// 
//   /**
//    * Channel is not open (default status)
//    */
//   public static final int CLOSED    = 0;
// 
//   /**
//    * Channel is in the process of closing down
//    */
//   public static final int CLOSING   = 1;
// 
//   /**
//    * Channel is in the process of opening
//    */
//   public static final int OPENING   = 2;
// 
//   /**
//    * Channel is active and Events are flowing through it
//    */
//   public static final int FLOWING   = 3;
// 
//   /**
//    * Channel is in an error state and Events are not flowing through it. This state could be set
//    * during any operation: open, close, append, etc.
//    */
//   public static final int ERROR     = -1;
// 
//   public int getStatus();
// 
// }
// 
// Path: src/main/java/net/pixelcop/sewer/Sink.java
// public abstract class Sink extends Plumbing {
// 
//   private static final Logger LOG = LoggerFactory.getLogger(Sink.class);
// 
//   protected Sink subSink = null;
// 
//   /**
//    * Append an Event to the Sink
//    * @param event
//    * @throws IOException
//    */
//   public abstract void append(Event event) throws IOException;
// 
//   public void setSubSink(Sink subSink) {
//     this.subSink = subSink;
//   }
// 
//   public Sink getSubSink() {
//     return subSink;
//   }
// 
//   /**
//    * Create the SubSink, if possible
//    *
//    * @return True if a SubSink was created, false otherwise.
//    */
//   public boolean createSubSink() {
//     if (getSinkFactory() == null) {
//       return false;
//     }
// 
//     subSink = getSinkFactory().build();
//     return true;
//   }
// 
//   @Override
//   protected void finalize() throws Throwable {
//     if (getStatus() != CLOSED) {
//       try {
//         close();
//       } catch (Throwable t) {
//         LOG.warn("Caught during finalizer close(): " + t.getMessage(), t);
//         // rethrow? this should be good enough since the object was disposed of anyway
//       }
//     }
//   }
// 
// }
// 
// Path: src/main/java/net/pixelcop/sewer/StatusProvider.java
// public interface StatusProvider {
// 
//   /**
//    * Channel is not open (default status)
//    */
//   public static final int CLOSED    = 0;
// 
//   /**
//    * Channel is in the process of closing down
//    */
//   public static final int CLOSING   = 1;
// 
//   /**
//    * Channel is in the process of opening
//    */
//   public static final int OPENING   = 2;
// 
//   /**
//    * Channel is active and Events are flowing through it
//    */
//   public static final int FLOWING   = 3;
// 
//   /**
//    * Channel is in an error state and Events are not flowing through it. This state could be set
//    * during any operation: open, close, append, etc.
//    */
//   public static final int ERROR     = -1;
// 
//   public int getStatus();
// 
// }


// Path: src/main/java/net/pixelcop/sewer/source/TCPReaderThread.java
import static net.pixelcop.sewer.StatusProvider.*;
import java.io.BufferedInputStream;
import java.io.DataInputStream;
import java.io.EOFException;
import java.io.IOException;
import java.net.Socket;
import net.pixelcop.sewer.Sink;
import net.pixelcop.sewer.StatusProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

package net.pixelcop.sewer.source;





public abstract class TCPReaderThread extends Thread {

  private static final Logger LOG = LoggerFactory.getLogger(TCPReaderThread.class);

  protected final Socket socket;
  protected final Sink sink;
  protected DataInputStream in;
  protected final StatusProvider statusProvider;

  public TCPReaderThread(String name, Socket socket, Sink sink, StatusProvider statusProvider) {