// Path: src/main/java/net/pixelcop/sewer/ByteArrayEvent.java
// public class ByteArrayEvent implements Event {
// 
//   private static final String NULL = "null";
// 
//   private byte[] body; // TODO use re-usable buffer for reading sequence files
// 
//   public ByteArrayEvent() {
//   }
// 
//   public ByteArrayEvent(byte[] body) {
//     this.body = body;
//   }
// 
//   public ByteArrayEvent(DataInput in) throws IOException {
//     readFields(in);
//   }
// 
//   @Override
//   public void write(DataOutput out) throws IOException {
//     out.writeInt(body.length);
//     out.write(body);
//   }
// 
//   @Override
//   public void readFields(DataInput in) throws IOException {
//     this.body = new byte[in.readInt()];
//     in.readFully(body);
//   }
// 
//   @Override
//   public String toString() {
//     if (body == null) {
//       return NULL;
//     }
//     return new String(body);
//   }
// 
// }
// 
// 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/Source.java
// public abstract class Source extends Plumbing {
// 
//   /**
//    * The class which this source uses for new Events
//    *
//    * @return
//    */
//   public abstract Class<?> getEventClass();
// 
// }


import java.io.IOException;
import java.net.Socket;
import net.pixelcop.sewer.ByteArrayEvent;
import net.pixelcop.sewer.Sink;
import net.pixelcop.sewer.Source;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

          @Override
          public void read() throws IOException {
            ByteArrayEvent event = new ByteArrayEvent(in);
            this.sink.append(event);
          }

        };

      }
    };

    this.serverThread.start();
  }

  @Override
  public void close() throws IOException {
    setStatus(CLOSING);
    LOG.info("Closing " + this.getClass().getSimpleName());
    try {
      LOG.debug("joining server thread");
      this.serverThread.join();
      LOG.debug("server thread has joined");
    } catch (InterruptedException e) {
      LOG.error("Interrupted waiting for server thread to join");
    }
    try {
      this.serverThread.joinReaders();
      LOG.debug("all reader threads have joined");
    } catch (InterruptedException e) {
// Path: src/main/java/net/pixelcop/sewer/ByteArrayEvent.java
// public class ByteArrayEvent implements Event {
// 
//   private static final String NULL = "null";
// 
//   private byte[] body; // TODO use re-usable buffer for reading sequence files
// 
//   public ByteArrayEvent() {
//   }
// 
//   public ByteArrayEvent(byte[] body) {
//     this.body = body;
//   }
// 
//   public ByteArrayEvent(DataInput in) throws IOException {
//     readFields(in);
//   }
// 
//   @Override
//   public void write(DataOutput out) throws IOException {
//     out.writeInt(body.length);
//     out.write(body);
//   }
// 
//   @Override
//   public void readFields(DataInput in) throws IOException {
//     this.body = new byte[in.readInt()];
//     in.readFully(body);
//   }
// 
//   @Override
//   public String toString() {
//     if (body == null) {
//       return NULL;
//     }
//     return new String(body);
//   }
// 
// }
// 
// 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/Source.java
// public abstract class Source extends Plumbing {
// 
//   /**
//    * The class which this source uses for new Events
//    *
//    * @return
//    */
//   public abstract Class<?> getEventClass();
// 
// }


// Path: src/main/java/net/pixelcop/sewer/source/TcpWriteableEventSource.java
import java.io.IOException;
import java.net.Socket;
import net.pixelcop.sewer.ByteArrayEvent;
import net.pixelcop.sewer.Sink;
import net.pixelcop.sewer.Source;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


          @Override
          public void read() throws IOException {
            ByteArrayEvent event = new ByteArrayEvent(in);
            this.sink.append(event);
          }

        };

      }
    };

    this.serverThread.start();
  }

  @Override
  public void close() throws IOException {
    setStatus(CLOSING);
    LOG.info("Closing " + this.getClass().getSimpleName());
    try {
      LOG.debug("joining server thread");
      this.serverThread.join();
      LOG.debug("server thread has joined");
    } catch (InterruptedException e) {
      LOG.error("Interrupted waiting for server thread to join");
    }
    try {
      this.serverThread.joinReaders();
      LOG.debug("all reader threads have joined");
    } catch (InterruptedException e) {
      LOG.error("Interrupted waiting for reader threads to join");