public class AvroEventSource extends EventSource.Base {
  /*
   * In this version I am setting the following constants same as for the thrift
   * case. Seems like these constants don't really need to depend on the
   * underlying implementation, so maybe we can give them more general names
   * later.
   */
  final static int DEFAULT_QUEUE_SIZE = FlumeConfiguration.get()
      .getThriftQueueSize();
  final static long MAX_CLOSE_SLEEP = FlumeConfiguration.get()
      .getThriftCloseMaxSleep();

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

  public static final String C_TRUNCATE = "truncate";
  public static final String A_SERVERPORT = "serverPort";
  public static final String A_QUEUE_CAPACITY = "queueCapacity";
  public static final String A_QUEUE_FREE = "queueFree";
  public static final String A_ENQUEUED = "enqueued";
  public static final String A_DEQUEUED = "dequeued";
  // BytesIN in here (unlike the Thrift version) corresponds to the total bytes
  // of Event.body shipped.
  public static final String A_BYTES_IN = "bytesIn";
  final int port;
  private FlumeEventAvroServerImpl svr;
  final BlockingQueue<Event> q;
  final AtomicLong enqueued = new AtomicLong();
  final AtomicLong dequeued = new AtomicLong();
  final AtomicLong bytesIn = new AtomicLong();
  final boolean shouldTruncate;
  boolean closed = true;

  /**
   * Create a Avro event source listening on port with a qsize buffer.
   */
  public AvroEventSource(int port, int qsize, boolean truncates) {
    this.port = port;
    this.svr = new FlumeEventAvroServerImpl(port);
    this.q = new LinkedBlockingQueue<Event>(qsize);
    this.shouldTruncate = truncates;
  }

  /**
   * Get reportable data from the Avro event source.
   */
  @Override
  synchronized public ReportEvent getMetrics() {
    ReportEvent rpt = super.getMetrics();
    rpt.setLongMetric(A_SERVERPORT, port);
    rpt.setLongMetric(A_QUEUE_CAPACITY, q.size());
    rpt.setLongMetric(A_QUEUE_FREE, q.remainingCapacity());
    rpt.setLongMetric(A_ENQUEUED, enqueued.get());
    rpt.setLongMetric(A_DEQUEUED, dequeued.get());
    rpt.setLongMetric(A_BYTES_IN, bytesIn.get());
    return rpt;
  }

  /**
   * This constructor allows the for an arbitrary blocking queue implementation.
   */
  public AvroEventSource(int port, BlockingQueue<Event> q, boolean truncates) {
    Preconditions.checkNotNull(q);
    this.port = port;
    this.q = q;
    this.shouldTruncate = truncates;
  }

  public AvroEventSource(int port) {
    this(port, DEFAULT_QUEUE_SIZE, false);
  }

  /**
   * Exposed for testing.
   */
  void enqueue(Event e) throws IOException {
    try {
      q.put(e);
      enqueued.getAndIncrement();
      bytesIn.getAndAdd(e.getBody().length);
    } catch (InterruptedException e1) {
      LOG.error("blocked append was interrupted", e1);
      throw new IOException(e1);
    }
  }

  /**
   * {@inheritDoc}
   */
  @Override
  synchronized public void open() throws IOException {

    this.svr = new FlumeEventAvroServerImpl(port) {
      @Override
      public void append(AvroFlumeEvent evt) {
        // convert AvroEvent evt -> e
        Event e = AvroEventConvertUtil.toFlumeEvent(evt, shouldTruncate);
        try {
          enqueue(e);
        } catch (IOException e1) {
          e1.printStackTrace();
        }
        super.append(evt);
      }
    };
    LOG.info(String.format("Avro listening server on port %d...", port));
    this.svr.start();
    this.closed = false;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  synchronized public void close() throws IOException {

    long sz = q.size();
    LOG.info(String.format("Queue still has %d elements ...", sz));

    // Close down the server
    this.svr.close();

    // drain the queue
    // TODO (jon) parameterize queue drain max sleep is one minute
    long maxSleep = MAX_CLOSE_SLEEP;
    long start = Clock.unixTime();
    while (q.peek() != null) {
      if (Clock.unixTime() - start > maxSleep) {
        if (sz == q.size()) {
          // no progress made, timeout and close it.
          LOG.warn("Close timed out due to no progress.  Closing despite having "
              + q.size() + " values still enqueued");
          return;
        }
        // there was some progress, go another cycle.
        start = Clock.unixTime();
      }

      try {
        Thread.sleep(100);
      } catch (InterruptedException e) {
        LOG.error("Unexpected interrupt of close " + e.getMessage(), e);
        Thread.currentThread().interrupt();
        closed = true;
        throw new IOException(e);
      }
    }

    closed = true;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public Event next() throws IOException {
    try {
      Event e = null;
      // block until an event shows up
      while ((e = q.poll(100, TimeUnit.MILLISECONDS)) == null) {

        synchronized (this) {
          // or bail out if closed
          if (closed) {
            return null;
          }
        }
      }
      // return the event
      synchronized (this) {
        dequeued.getAndIncrement();
        updateEventProcessingStats(e);
        return e;
      }
    } catch (InterruptedException e) {
      throw new IOException("Waiting for queue element was interrupted! "
          + e.getMessage(), e);
    }
  }

  public static SourceBuilder builder() {
    return new SourceBuilder() {
      @Override
      public EventSource build(Context ctx, String... argv) {
        Preconditions.checkArgument(argv.length == 1,
            "usage: avroSource(port{," + C_TRUNCATE + "=false})");
        int port = Integer.parseInt(argv[0]);

        String val = ctx.getObj(C_TRUNCATE, String.class);
        boolean truncates = (val == null) ? false : Boolean.parseBoolean(val);

        return new AvroEventSource(port, DEFAULT_QUEUE_SIZE, truncates);
      }
    };
  }
} public class SynthSourceRndSize extends EventSource.Base {
  static final Logger LOG = LoggerFactory.getLogger(SynthSourceRndSize.class);

  final long total;
  final long seed;
  final Random rand;
  long count = 0;
  final int minBodySize;
  final int maxBodySize;

  public SynthSourceRndSize(long total, int lowLimit, int upLimit) {
    this(total, lowLimit, upLimit, Clock.unixTime());

  }

  public SynthSourceRndSize(long total, int minsize, int maxsize, long seed) {
    this.rand = new Random(seed);
    this.seed = seed;
    this.minBodySize = minsize;
    this.maxBodySize = maxsize;
    this.total = count;
  }

  @Override
  public void close() throws IOException {
    LOG.info("closing SynthSourceRandSize");
  }

  @Override
  public Event next() throws IOException {
    if (count >= total && total != 0)
      return null;// end marker if gotten to count

    int range = this.maxBodySize - this.minBodySize;
    int size = this.minBodySize + (range > 0 ? this.rand.nextInt(range) : 0);
    count++;
    byte[] data = new byte[size];
    rand.nextBytes(data);
    Event e = new EventImpl(data);
    updateEventProcessingStats(e);
    return e;
  }

  @Override
  public void open() throws IOException {
    LOG.info("Resetting count and seed; openingSynthSourceRandSize");
    // resetting the seed
    count = 0;
    rand.setSeed(seed);
  }

  public static SourceBuilder builder() {
    return new SourceBuilder() {
      @Override
      public EventSource build(Context ctx, String... argv) {
        int minSize = 0;
        int maxSize = 10;
        long total = 0;

        if (argv.length != 3) {
          throw new IllegalArgumentException(
              "usage: synthrndsize(count=0, minsize=0, maxsize=10) // count=0 infinite");
        }
        total = Long.parseLong(argv[0]);
        minSize = Integer.parseInt(argv[1]);
        maxSize = Integer.parseInt(argv[2]);
        return new SynthSourceRndSize(total, minSize, maxSize);
      }
    };
  }
} public class ExecNioSource extends EventSource.Base {
  // What command to run
  final String command;
  // Should we restart the script when it finishes?
  final boolean restart;
  // Time to wait if restart is true
  final int period;
  // Return line by line, or aggregate into a single pair of events?
  final boolean inAggregateMode;

  // Flags used to signal the end of an input stream
  private final AtomicBoolean errFinished = new AtomicBoolean(false);
  private final AtomicBoolean outFinished = new AtomicBoolean(false);

  private final BlockingQueue<Event> eventQueue = new LinkedBlockingQueue<Event>(
      1000);

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

  public static final String A_PROC_SOURCE = "procsource";
  public static final String A_EXEC_CMD = "execcmd";

  // Input sources
  private ReadableByteChannel stdout = null;
  private ReadableByteChannel stderr = null;

  // Two threads to read from each source
  ReaderThread readOut = null, readErr = null;
  private InputStreamPipe stdinISP = null, stderrISP = null;

  private Process proc = null;

  // Used to signal that both reader and err threads have exited
  private CountDownLatch latch = new CountDownLatch(2);

  /**
   * 
   * @param command
   *          Command line to exec and get output from.
   * @param aggregate
   *          if true, return all the data from a single exec, if false, return
   *          an event per line.
   * @param restart
   *          if true, restart exec every period ms after previous exit. if
   *          false, have source return done status after execution.
   * @param period
   *          milliseconds to wait after exec exists before restarting if
   *          restart is true.
   */
  ExecNioSource(String command, boolean aggregate, boolean restart, int period) {
    this.command = command;
    this.inAggregateMode = aggregate;
    this.restart = restart;
    this.period = period;
  }

  /**
   * Create an event gathered from an exec of a program.
   */
  static Event buildExecEvent(byte[] body, String tag, String command)
      throws InterruptedException {
    Event e = new EventImpl(body);
    Attributes.setString(e, A_PROC_SOURCE, tag);
    Attributes.setString(e, A_EXEC_CMD, command);
    Attributes.setString(e, Event.A_SERVICE, "exec");
    return e;
  }

  /**
   * Makes events from the supplied byte buffer and puts them into the specified
   * BlockingQueue. If it doesn't end with \n, then compact to shift the
   * leftover bytes to be beginning of the buffer. When this function exits, the
   * buffer is in write mode, still contains any leftovers and leaves position
   * pointing to the end of the incomplete line.
   * 
   * @param buf
   *          ByteBuffer in write mode. If this method exits normally, there are
   *          no remaining '\n's in the buffer and the buf is in write mode
   * @return true if any bytes have been consumed
   */
  static boolean extractLines(ByteBuffer buf, String command, String tag,
      BlockingQueue<Event> sync) throws InterruptedException {
    buf.flip();
    boolean madeProgress = false;
    int start = buf.position();
    buf.mark();
    while (buf.hasRemaining()) {
      byte b = buf.get();
      // TODO windows: ('\r\n') line separators
      if (b == '\n') {
        int end = buf.position();
        int sz = end - start - 1; // exclude '\n'
        int maxEventSz = (int) FlumeConfiguration.get().getEventMaxSizeBytes();
        if (sz > maxEventSz) {
          // Truncating path.
          byte[] body = new byte[(int) maxEventSz];
          buf.reset(); // go back to mark
          buf.get(body, 0, maxEventSz); // read data
          buf.position(end);
          buf.mark(); // new mark.
          start = buf.position();

          sync.put(buildExecEvent(body, tag, command));
          madeProgress = true;
        } else {
          byte[] body = new byte[sz];
          buf.reset(); // go back to mark
          buf.get(body, 0, sz); // read data
          buf.get(); // read '\n'
          buf.mark(); // new mark.
          start = buf.position();

          sync.put(buildExecEvent(body, tag, command));
          madeProgress = true;
        }
      }
    }

    // rewind for any left overs
    buf.reset();
    buf.compact(); // shift leftovers to front.
    return madeProgress;
  }

  /**
   * @param in
   *          byte buffer in write mode
   * @return true if needs to stay in drop mode, false if reached a '\n' char.
   *         Buffer is in write mode when it exits.
   */
  static boolean dropUntilNewLine(ByteBuffer in) {
    in.flip();

    while (in.hasRemaining()) {
      if (in.get() == '\n') {
        in.compact(); // get rid of everything and flip back into normal mode
        return false;
      }
    }
    // wipe out the data and stay in drop mode.
    in.clear();
    return true;
  }

  /**
   * Polls an input and formats lines read as events, places them on the event
   * queue.
   */
  class ReaderThread extends Thread {
    ReadableByteChannel readChan = null;
    volatile boolean shutdown = false;
    String tag;
    AtomicBoolean signalDone;

    ReaderThread(ReadableByteChannel input, String tag, AtomicBoolean signal) {
      super("ReaderThread (" + command + "-" + tag + ")");
      Preconditions.checkArgument(input != null);
      Preconditions.checkArgument(signal != null);
      this.readChan = input;
      this.tag = tag;
      this.signalDone = signal;
    }

    /**
     * Returns true of the process is terminated
     * 
     * @param proc
     * @return true if process completed, false if was in weird state
     */
    boolean isProcDone(Process proc) {
      try {
        proc.exitValue();
        return true;
      } catch (IllegalThreadStateException e) {
        // This is gross but the only java way to figure out if the
        // subprocess is running.
        return false;
      }
    }

    /**
     * Takes exec output and converts individual lines into events.
     */
    void doLineMode() {
      // make sure we have a buffer big enough to get relevant data.
      int maxEventSize = (int) FlumeConfiguration.get().getEventMaxSizeBytes();
      int bufSize = Math.max(Short.MAX_VALUE, maxEventSize * 4);
      try {
        ByteBuffer in = ByteBuffer.allocate(bufSize);
        boolean dropMode = false; // for truncations of extremely long lines

        while (!shutdown) {
          // If interrupted, this throws an IOException
          int read = readChan.read(in);

          if (read == 0) {
            // don't burn the cpu if nothing is read.
            Clock.sleep(10);
            continue;
          }

          if (read < 0) {
            // end of input stream reached.
            if (isProcDone(proc)) {
              shutdown = true;
            }
            continue;
          }

          // At this point, I have read data.
          if (dropMode) {
            dropMode = dropUntilNewLine(in);
            if (dropMode) {
              // didn't reach new line, keep dropping
              continue;
            }
            // fall through and do extract lines
          }

          // exits with 'in' in write mode
          extractLines(in, command, tag, eventQueue);

          // the leftovers bytes ideally should always be smaller than
          // maxEventSize, and has the invariant of not having a '\n' in it.
          if (in.position() > maxEventSize) {
            // read up to max size, and then throw out the rest
            LOG.error("Entry too long, truncating: " + in.position() + " > "
                + maxEventSize + "(max event size)");

            in.flip(); // read mode.
            byte[] buf = new byte[maxEventSize];
            in.get(buf);
            eventQueue.put(buildExecEvent(buf, command, tag));

            // We can now drop the remaining data, and continue to drop
            in.clear(); // back in write mode.
            dropMode = true;
          }
        }
      } catch (InterruptedException e) {
        // interruptions are only expected in shutdown.
        if (!shutdown) {
          LOG.warn(tag + " ReaderThread received "
              + "unexpected InterruptedException", e);
        }
      } catch (BufferOverflowException b) {
        // This should never happen (buffer is bigger than max event size)
        LOG.error("Event was too large for buffer", b);
      } catch (IOException e) {
        if (!shutdown) {
          LOG.warn(tag + " ReaderThread received unexpected IOException", e);
        }
      } finally {
        try {
          readChan.close();
        } catch (IOException i) {
          LOG.warn("Failed to close input stream in ExecNioSource", i);
        }
      }
      signalDone.set(true);
      latch.countDown();
    }

    /**
     * This takes exec and reads as much as it can from a single exec and then
     * creates a single event.
     */
    void doAggregateMode() {
      // make sure we have a buffer big enough to get relevant data.
      int maxEventSize = (int) FlumeConfiguration.get().getEventMaxSizeBytes();
      try {
        // TODO evaluate allocate vs allocateDirect
        ByteBuffer in = ByteBuffer.allocate(maxEventSize);

        Event evt = null;
        boolean dropMode = false;
        while (!shutdown) {
          // If interrupted, this throws an IOException
          int read = readChan.read(in);

          if (read == 0) { // nothing read? chill out for a bit.
            Clock.sleep(100);
            continue;
          }

          if (read < 0) { // read end of input stream.
            if (isProcDone(proc)) {
              shutdown = true;
            }

            if (evt != null) {
              // full event already saved off
              eventQueue.put(evt);
              break;
            }

            if (evt == null && in.position() != 0) {
              // event that didn't fill the buffer
              byte[] eventBuf = new byte[in.position()];
              in.flip();
              in.get(eventBuf);
              evt = buildExecEvent(eventBuf, command, tag);
              eventQueue.put(evt);
              break;
            }

            // buffer of 0 size, (didn't read anything): do nothing.
            break;
          }

          // if we read data, keep reading into buffer. If the buffer is full,
          // read the data out of the buf
          if (dropMode) {
            in.clear();
            continue;
          }
          if (in.remaining() == 0) {
            byte[] eventBuf = new byte[in.position()];
            in.flip();
            in.get(eventBuf);
            evt = buildExecEvent(eventBuf, command, tag);

            in.clear(); // don't need the data any more
            dropMode = true;
          }

        }
      } catch (InterruptedException e) {
        // interruptions are expected in shutdown.
        if (!shutdown) {
          LOG.warn(tag + " ReaderThread received "
              + "unexpected InterruptedException", e);
        }
      } catch (IOException e) {
        if (!shutdown) {
          LOG.warn(tag + " ReaderThread received unexpected IOException", e);
        }
      } finally {
        try {
          readChan.close();
        } catch (IOException i) {
          LOG.warn("Failed to close input stream in ExecEventSource", i);
        }
        signalDone.set(true);
        latch.countDown();
      }
    }

    /**
     * Blocks on a line of input to be available from an input stream; formats
     * as an event and then places it on a queue.
     */
    public void run() {
      if (inAggregateMode) {
        // are stderr and stdout different events? so this can do two events?
        doAggregateMode();
      } else {
        doLineMode();
      }
    }

    void shutdown() throws IOException {
      this.shutdown = true;
      if (this.readChan != null) {
        readChan.close();
        this.interrupt();
      }
    }
  }

  public void close() throws IOException {
    // Note that this does not guarantee that any further next() calls will
    // return the EOF null that signals the process shut down.

    readOut.shutdown();
    readErr.shutdown();
    boolean latched = false;
    // Want to make sure that both threads have exited before we kill the
    // process
    try {
      latched = latch.await(5000, TimeUnit.MILLISECONDS);
    } catch (InterruptedException e) {
      LOG.debug("Waiting for exec thread exit was interrupted", e);
    }

    stdinISP.shutdown();
    stderrISP.shutdown();
    if (proc != null) {
      proc.destroy();
      proc = null;
    }

    if (!latched) {
      throw new IOException("Timeout waiting for exec threads to exit");
    }
  }

  /**
   * Blocks on either output from stdout / stderr or process exit (at which
   * point it throws an exception)
   * 
   * @return an Event with two tags: the stream which produced the line
   */
  public Event next() throws IOException {
    Event line = null;
    while (true) {
      try {
        line = eventQueue.poll(1000, TimeUnit.MILLISECONDS);
        if (line == null) {
          if (errFinished.get() && outFinished.get()) {
            // We may have missed events between waking up and testing
            line = eventQueue.poll();
            if (line != null) {
              updateEventProcessingStats(line);
              return line;
            }
            if (restart) {
              close();
              Thread.sleep(period);
              open();
            } else {
              return null;
            }
          }
        } else {
          updateEventProcessingStats(line);
          return line;
        }
      } catch (InterruptedException e) {
        // this is in the driver so needs to re-flag interrupted status
        LOG.warn("Exec next was interrupted " + e);
        Thread.currentThread().interrupt();
        throw new RuntimeException("ExecEventSource was interrupted - " + e);
      }
    }
  }

  /**
   * Starts a Process and two threads to read from stdout / stderr
   */
  public void open() throws IOException {
    if (proc != null) {
      throw new IllegalStateException("Tried to open exec process twice");
    }
    latch = new CountDownLatch(2);
    outFinished.set(false);
    errFinished.set(false);
    proc = Runtime.getRuntime().exec(command);

    // Just reading from stdout and stderr can block, so we wrap them with
    // InputStreamPipe allows them to be nonblocking.
    stdinISP = new InputStreamPipe(proc.getInputStream());
    stderrISP = new InputStreamPipe(proc.getErrorStream());
    stdout = (ReadableByteChannel) stdinISP.getChannel();
    stderr = (ReadableByteChannel) stderrISP.getChannel();

    readOut = new ReaderThread(stdout, "STDOUT", outFinished);
    readErr = new ReaderThread(stderr, "STDERR", errFinished);

    stdinISP.start();
    stderrISP.start();
    readOut.start();
    readErr.start();
  }

  protected static class Builder extends SourceBuilder {
    /**
     * Takes 1-4 arguments - the command to run, whether to aggregate each
     * output as a single event, whether to restart after one execution is
     * finished, and how often if so to restart.
     */
    @Override
    public EventSource build(Context ctx, String... argv) {
      Preconditions.checkArgument(argv.length >= 1 && argv.length <= 4,
          "exec(\"cmdline \"[,aggregate [,restart [,period]]]], )");
      String command = argv[0];
      boolean aggregate = false;
      boolean restart = false;
      int period = 0;
      if (argv.length >= 2) {
        aggregate = Boolean.parseBoolean(argv[1]);
      }
      if (argv.length >= 3) {
        restart = Boolean.parseBoolean(argv[2]);
      }
      if (argv.length >= 4) {
        period = Integer.parseInt(argv[3]);
      }
      return new ExecNioSource(command, aggregate, restart, period);
    }
  }

  /**
   * This builder creates a source that periodically execs a program and takes
   * the entire output as the body of a event. It takes two arguments - the
   * command to run, and a time period to sleep in millis before executing
   * again.
   */
  public static SourceBuilder buildPeriodic() {
    return new SourceBuilder() {
      @Override
      public EventSource build(Context ctx, String... argv) {
        Preconditions.checkArgument(argv.length == 2,
            "execPeriodic(\"cmdline \",period)");
        String command = argv[0];
        boolean aggregate = true;
        boolean restart = true;
        int period = Integer.parseInt(argv[1]);
        return new ExecNioSource(command, aggregate, restart, period);
      }
    };
  }

  /**
   * This builder creates a source that execs a long running program and takes
   * each line of input as the body of an event. It takes one argument, the
   * command to run. If the command exits, the exec source returns null signally
   * end of records.
   */
  public static SourceBuilder buildStream() {
    return new SourceBuilder() {
      @Override
      public EventSource build(Context ctx, String... argv) {
        Preconditions.checkArgument(argv.length == 1,
            "execStream(\"cmdline \")");
        String command = argv[0];
        boolean aggregate = false;
        boolean restart = false;
        int period = 0;
        return new ExecNioSource(command, aggregate, restart, period);
      }
    };
  }

  public static SourceBuilder builder() {
    return new Builder();
  }
} public class SeqfileEventSource extends EventSource.Base {
  static final Logger LOG = LoggerFactory.getLogger(SeqfileEventSource.class);

  private String fname;
  private SequenceFile.Reader reader;

  public SeqfileEventSource(String fname) {
    this.fname = fname;
  }

  /**
   * We assume that the file is local.
   */
  static public SeqfileEventSource openLocal(String fname) throws IOException {
    SeqfileEventSource src = new SeqfileEventSource(fname);
    src.open();
    return src;
  }

  public Event next() throws IOException {
    Preconditions.checkNotNull(reader);
    WriteableEventKey k = new WriteableEventKey();
    WriteableEvent e = new WriteableEvent();

    boolean ok = reader.next(k, e);

    if (!ok)
      return null;

    Event evt = e.getEvent();
    updateEventProcessingStats(evt);
    return evt;
  }

  @Override
  public void close() throws IOException {
    LOG.debug("closing SeqfileEventSource " + fname);
    if (reader == null) {
      return;
    }
    reader.close();
    reader = null;
  }

  @Override
  public void open() throws IOException {
    LOG.debug("opening SeqfileEventSource " + fname);
    Configuration conf = FlumeConfiguration.get();
    FileSystem fs = FileSystem.getLocal(conf);
    reader = new SequenceFile.Reader(fs, new Path(fname), conf);
  }

  @Override
  public String toString() {
    return "SeqfileEventSource name=" + fname;
  }

  public static SourceBuilder builder() {
    return new SourceBuilder() {

      @Override
      public EventSource build(Context ctx, String... argv) {
        if (argv.length != 1) {
          throw new IllegalArgumentException("usage: seqfile(filename)");
        }

        return new SeqfileEventSource(argv[0]);
      }

    };
  }

} public class IrcSource extends EventSource.Base {
  static final Logger LOG = LoggerFactory.getLogger(IrcSource.class);

  private IRCConnection conn;

  String host;
  int port;
  String nick;
  String pass;
  String user;
  String name;
  String chan;
  boolean ssl; // later iteraton

  BlockingQueue<Event> q = new LinkedBlockingQueue<Event>();

  public IrcSource(String host, int port, String nick, String pass,
      String user, String name, String chan, boolean ssl) {
    this.host = host;
    this.port = port;
    this.nick = nick;
    this.pass = pass;
    this.user = user;
    this.name = name;
    this.chan = chan;
    this.ssl = ssl;
  }

  public IrcSource(String host, int port, String nick, String chan) {
    this(host, port, nick, null, null, null, chan, false);
  }

  @Override
  public void open() throws IOException {
    // TODO (jon) check if the nick usage was successful or not
    conn = new IRCConnection(host, new int[] { port }, pass, nick, user, name);
    conn.addIRCEventListener(new Listener());
    conn.setEncoding(FlumeConfiguration.get().getFlurkerEncoding());
    conn.setPong(true);
    conn.setDaemon(false);
    conn.setColors(false);
    conn.connect();

    conn.send("join " + chan);
    // TODO(jon) check if channel join was successful or not
  }

  @Override
  public void close() throws IOException {
    conn.close();
  }

  private void append(String s) {
    if (s == null) {
      LOG.error("null append!");
      return;
    }
    q.add(new EventImpl(s.getBytes()));
  }

  // Need to pick which call backs should be "sources" and which are "sinks".

  /**
   * Treats IRC events. The most of them are just printed.
   * 
   * TODO (jon) since we have fields, we can use store this as structured data.
   */
  public class Listener implements IRCEventListener {

    public void onRegistered() {
      append("Connected");
    }

    public void onDisconnected() {
      append("Disconnected");
    }

    public void onError(String msg) {
      append("Error: " + msg);
    }

    public void onError(int num, String msg) {
      append("Error #" + num + ": " + msg);
    }

    public void onInvite(String chan, IRCUser u, String nickPass) {
      append(chan + "> " + u.getNick() + " invites " + nickPass);
    }

    public void onJoin(String chan, IRCUser u) {
      append(chan + "> " + u.getNick() + " joins");
    }

    public void onKick(String chan, IRCUser u, String nickPass, String msg) {
      append(chan + "> " + u.getNick() + " kicks " + nickPass);
    }

    public void onMode(IRCUser u, String nickPass, String mode) {
      append("Mode: " + u.getNick() + " sets modes " + mode + " " + nickPass);
    }

    public void onMode(String chan, IRCUser u, IRCModeParser mp) {
      append(chan + "> " + u.getNick() + " sets mode: " + mp.getLine());
    }

    public void onNick(IRCUser u, String nickNew) {
      append("Nick: " + u.getNick() + " is now known as " + nickNew);
    }

    public void onNotice(String target, IRCUser u, String msg) {
      append(target + "> " + u.getNick() + " (notice): " + msg);
    }

    public void onPart(String chan, IRCUser u, String msg) {
      append(chan + "> " + u.getNick() + " parts");
    }

    public void onPrivmsg(String chan, IRCUser u, String msg) {
      append(chan + "> " + u.getNick() + ": " + msg);
    }

    public void onQuit(IRCUser u, String msg) {
      append("Quit: " + u.getNick());
    }

    public void onReply(int num, String value, String msg) {
      append("Reply #" + num + ": " + value + " " + msg);
    }

    public void onTopic(String chan, IRCUser u, String topic) {
      append(chan + "> " + u.getNick() + " changes topic into: " + topic);
    }

    public void onPing(String p) {

    }

    public void unknown(String a, String b, String c, String d) {
      append("UNKNOWN: " + a + " b " + c + " " + d);
    }
  }

  @Override
  public Event next() throws IOException {
    try {
      Event e = q.take();
      updateEventProcessingStats(e);
      return e;
    } catch (InterruptedException e) {
      LOG.error("IrcSource interrupted", e);
      throw new IOException(e);
    }
  }

  public static SourceBuilder builder() {
    return new SourceBuilder() {
      @Override
      public EventSource build(Context ctx, String... argv) {
        Preconditions.checkArgument(argv.length == 4,
            "usage: ircSource(server, port, nick, chan)");
        String server = argv[0];
        int port = Integer.parseInt(argv[1]);
        String nick = argv[2];
        String chan = argv[3];
        return new IrcSource(server, port, nick, chan);
      }
    };
  }
} public class ScribeEventSource extends ThriftServer implements EventSource,
    scribe.Iface {
  static final Logger LOG = LoggerFactory.getLogger(ScribeEventSource.class);
  public static final String A_SERVERPORT = "serverPort";

  final BlockingQueue<Event> pendingQueue = new LinkedBlockingQueue<Event>();

  final static public String SCRIBE_CATEGORY = "scribe.category";
  final AtomicBoolean running = new AtomicBoolean(false);
  long startedTime = 0;
  int port = 0;

  final static Event DONE_EVENT = new EventImpl(new byte[0]);

  /**
   * Construct a scribe event source.
   * 
   * @param port
   *          port the server will listen on
   */
  public ScribeEventSource(int port) {
    // turn off thrift strict read & write (respectively), otw legacy
    // thrift clients (ie scribe clients) won't be able to connect. This
    // mimics what scribed does.
    super(false, false);

    this.port = port;
  }

  public ScribeEventSource() {
    this(FlumeConfiguration.DEFAULT_SCRIBE_SOURCE_PORT);
  }

  /**
   * Stops the Thrift server
   */
  @Override
  public synchronized void close() throws IOException {
    running.set(false);
    this.stop();
    pendingQueue.add(DONE_EVENT);
  }

  /**
   * Blocks until a new event is available.
   */
  @Override
  public Event next() throws IOException {
    try {
      Event e = pendingQueue.take();
      if (e == DONE_EVENT) {
        return null;
      }
      return e;
    } catch (InterruptedException e) {
      LOG.error("ScribeEventSource was interrupted while waiting for an event",
          e);
      throw new IOException(e);
    }
  }

  /**
   * Starts a Thrift server and waits for it to come up.
   */
  @Override
  public synchronized void open() throws IOException {
    try {
      // Start the thrift server with a framed transport - suitable for
      // scribe clients
      this.start(new scribe.Processor(this), "ScribeEventSource",
          new TSaneServerSocket(port) {
            // we are providing the transport to ThriftServer -- the sole
            // job of this sane server subclass is to wrap the socket
            // with a framed transport
            protected TTransport acceptImpl() throws TTransportException {
              return new TFramedTransport(super.acceptImpl());
            }
          });
      running.set(true);
      startedTime = Clock.unixTime();
    } catch (TTransportException e) {
      LOG.error("Could not start Thrift Scribe server", e);
      throw new IOException(e);
    }
  }

  /**
   * This is the only API call required for scribe
   */
  @Override
  public ResultCode Log(List<LogEntry> messages) throws TException {
    // This will only happen if the log call arrives between
    // the Thrift socket opening and serve(...) returning - a small window!
    if (!running.get()) {
      return ResultCode.TRY_LATER;
    }
    for (LogEntry l : messages) {
      EventImpl e = new EventImpl(l.message.getBytes());
      e.set(SCRIBE_CATEGORY, l.category.getBytes());
      pendingQueue.add(e);
    }
    return ResultCode.OK;
  }

  /*
   * All following methods are required for fb303
   */

  @Override
  public long aliveSince() throws TException {
    return startedTime;
  }

  @Override
  public long getCounter(String key) throws TException {
    throw new TException("getCounter not implemented!");
  }

  @Override
  public Map<String, Long> getCounters() throws TException {
    throw new TException("getCounters not implemented!");
  }

  @Override
  public String getCpuProfile(int profileDurationInSec) throws TException {
    throw new TException("getCpuProfile not implemented!");
  }

  @Override
  public String getName() {
    return "Flume Scribe Event Server";
  }

  @Override
  public String getOption(String key) throws TException {
    throw new TException("getOption not implemented!");
  }

  @Override
  public Map<String, String> getOptions() throws TException {
    throw new TException("getOptions not implemented!");
  }

  @Override
  public fb_status getStatus() throws TException {
    return running.get() ? fb_status.ALIVE : fb_status.STOPPED;
  }

  @Override
  public String getStatusDetails() throws TException {
    throw new TException("getStatusDetails not implemented!");
  }

  @Override
  public String getVersion() throws TException {
    return VersionInfo.getVersion();
  }

  @Override
  public void reinitialize() throws TException {
    throw new TException("reinitialize not implemented!");
  }

  @Override
  public void setOption(String key, String value) throws TException {
    throw new TException("setOption not implemented!");
  }

  @Override
  public void shutdown() throws TException {
    throw new TException("shutdown not implemented!");
  }

  /**
   * Builder takes one optional argument: the port to start on
   */
  public static SourceBuilder builder() {
    return new SourceBuilder() {
      @Override
      public EventSource build(Context ctx, String... argv) {
        Preconditions.checkArgument(argv.length <= 1, "usage: scribe[(port={"
            + FlumeConfiguration.DEFAULT_SCRIBE_SOURCE_PORT + "})]");
        int port = FlumeConfiguration.get().getScribeSourcePort();
        if (argv.length >= 1) {
          port = Integer.parseInt(argv[0]);
        }
        return new ScribeEventSource(port);
      }
    };
  }

  @Override
  public ReportEvent getMetrics() {
    // TODO(henry): add metrics
    // TODO missing EventSource stats
    ReportEvent rpt = new ReportEvent("scribe-source");
    rpt.setLongMetric(A_SERVERPORT, port);
    return rpt;
  }

  @Override
  public Map<String, Reportable> getSubMetrics() {
    return ReportUtil.noChildren();
  }

  @Override
  public void getReports(String namePrefix, Map<String, ReportEvent> reports) {
    reports.put(namePrefix + getName(), getMetrics());
  }
} public class SyslogTcpSourceThreads extends EventSource.Base {
  static final Logger LOG = LoggerFactory
      .getLogger(SyslogTcpSourceThreads.class);

  final public static int SYSLOG_TCP_PORT = 514;
  final int port;
  final LinkedBlockingQueue<Event> eventsQ = new LinkedBlockingQueue<Event>(100000);
  final List<ReaderThread> readers = Collections
      .synchronizedList(new ArrayList<ReaderThread>());
  final AtomicLong rejects = new AtomicLong();
  volatile boolean closed = true;

  public SyslogTcpSourceThreads(int port) {
    this.port = port;
  }

  public SyslogTcpSourceThreads() {
    this(SYSLOG_TCP_PORT); // this is syslog-ng's default tcp port.
  }

  // must synchronize this variable
  ServerThread svrthread;
  ServerSocket sock = null;
  Object sockLock = new Object();

  /**
   * This thread just waits to accept incoming connections and spawn a reader
   * thread.
   */
  class ServerThread extends Thread {
    final int port;

    ServerThread(int port) {
      this.port = port;
    }

    @Override
    public void run() {
      while (!closed) {
        ServerSocket mySock = null; // guarantee no NPE at accept
        synchronized (sockLock) {
          mySock = sock; // get a local reference to sock.
        }

        if (mySock == null || mySock.isClosed())
          return;

        try {
          Socket client = mySock.accept();
          new ReaderThread(client).start();
        } catch (IOException e) {
          if (!closed) {
            // could be IOException where we run out of file/socket handles.
            LOG.error("accept had a problem", e);
          }
          return;
        }
      }
    }
  };

  /**
   * This thread takes a accepted socket and pull data out until it is empty.
   */
  class ReaderThread extends Thread {
    Socket in;

    ReaderThread(Socket sock) {
      readers.add(this);
      this.in = sock;
    }

    @Override
    public void run() {
      try {
        // process this connection.
        DataInputStream dis = new DataInputStream(in.getInputStream());
        while (!closed) {
          try {
            Event e = SyslogWireExtractor.extractEvent(dis);
            if (e == null)
              break;
            eventsQ.put(e);
          } catch (EventExtractException ex) {
            rejects.incrementAndGet();
          }
        }
        // done.
        in.close();

      } catch (IOException e) {
        LOG.error("IOException with SyslogTcpSources", e);
      } catch (InterruptedException e1) {
        LOG.error("put into Queue interupted" + e1);
      } finally {
        if (in != null && in.isConnected()) {
          try {
            in.close();
          } catch (IOException e) {
            e.printStackTrace();
          }
        }
        readers.remove(this);
      }
    }
  }

  @Override
  public void close() throws IOException {
    LOG.info("Closing " + this);
    synchronized (sockLock) {
      closed = true;
      if (sock != null) {
        sock.close();
        sock = null;
      }
    }
    // wait for all readers to close (This is not robust!)
    if (readers.size() != 0) {
      List<ReaderThread> rs = new ArrayList<ReaderThread>(readers); // 
      for (ReaderThread r : rs) {
        try {
          r.join();
        } catch (InterruptedException e) {
          LOG.error("Reader threads interrupted, but we are closing", e);
        }
      }
    }
    try {
      if (svrthread != null) {
        svrthread.join();
        svrthread = null;
      }
    } catch (InterruptedException e) {
      LOG.error("Reader threads interrupted, but we are closing", e);
    }
  };

  @Override
  public Event next() throws IOException {
    Event e = null;
    try {
      while ((e = eventsQ.poll(1000, TimeUnit.MILLISECONDS)) == null && !closed) {
        // Do nothing, just checking variables if nothing has arrived.
      }

    } catch (InterruptedException e1) {
      LOG.error("Tcp source polling interrupted ", e1); // Fail by throwing exn
      throw new IOException(e1);
    }
    updateEventProcessingStats(e);
    return e;
  }

  @Override
  public void open() throws IOException {
    LOG.info("Opening " + this);
    synchronized (sockLock) {
      if (!closed) {
        throw new IOException("Attempted to double open socket");
      }
      closed = false;

      if (sock == null) {
        // depending on number of connections, may need to increase backlog
        // value (automatic server socket argument, default is 50)
        try {
          sock = new ServerSocket(port);
          sock.setReuseAddress(true);
        } catch (IOException e) {
          throw new IOException("failed to create serversocket " + e);
        }
      }
    }
    svrthread = new ServerThread(port);
    svrthread.start();
  }

  public static SourceBuilder builder() {
    return new SourceBuilder() {

      @Override
      public EventSource build(Context ctx, String... argv) {
        int port = SYSLOG_TCP_PORT; // default udp port, need root permissions
        // for this.
        if (argv.length > 1) {
          throw new IllegalArgumentException("usage: syslogTcp([port no]) ");
        }

        if (argv.length == 1) {
          port = Integer.parseInt(argv[0]);
        }

        return new SyslogTcpSourceThreads(port);
      }

    };
  }

} public class TailDirSource extends EventSource.Base {
  public static final Logger LOG = LoggerFactory.getLogger(TailDirSource.class);
  public static final String USAGE = "usage: tailDir(\"dirname\"[, fileregex=\".*\"[, startFromEnd=false[, recurseDepth=0]]])";
  private DirWatcher watcher;
  private ConcurrentMap<String, DirWatcher> subdirWatcherMap;
  private TailSource tail;
  final private File dir;
  final private String regex;
  private final boolean startFromEnd;
  private final int recurseDepth;

  final private String delimRegex;
  final private DelimMode delimMode;

  // Indicates whether dir was checked. It is false before source is open
  // and set to true after the first check of a dir
  private volatile boolean dirChecked = false;

  final private AtomicLong filesAdded = new AtomicLong();
  final private AtomicLong filesDeleted = new AtomicLong();
  final private AtomicLong subdirsAdded = new AtomicLong();
  final private AtomicLong subdirsDeleted = new AtomicLong();

  final public static String A_FILESADDED = "filesAdded";
  final public static String A_FILESDELETED = "filesDeleted";
  final public static String A_FILESPRESENT = "filesPresent";
  final public static String A_SUBDIRSADDED = "subdirsAdded";
  final public static String A_SUBDIRSDELETED = "subdirsDeleted";

  public TailDirSource(File f, String regex) {
    this(f, regex, false);
  }

  public TailDirSource(File f, String regex, boolean startFromEnd) {
    this(f, regex, startFromEnd, 0);
  }

  public TailDirSource(File f, String regex, boolean startFromEnd,
      int recurseDepth) {
    Preconditions.checkArgument(f != null, "File should not be null!");
    Preconditions.checkArgument(regex != null,
        "Regex filter should not be null");

    this.dir = f;
    this.regex = regex;
    this.startFromEnd = startFromEnd;
    this.recurseDepth = recurseDepth;

    // 100 ms between checks
    this.tail = new TailSource(100);
    this.delimMode = null;
    this.delimRegex = null;
  }

  public TailDirSource(File f, String regex, boolean startFromEnd,
      int recurseDepth, String delimRegex, DelimMode delimMode) {
    Preconditions.checkArgument(f != null, "File should not be null!");
    Preconditions.checkArgument(regex != null,
        "Regex filter should not be null");

    this.dir = f;
    this.regex = regex;
    this.startFromEnd = startFromEnd;
    this.recurseDepth = recurseDepth;

    // 100 ms between checks
    this.tail = new TailSource(100);
    this.delimRegex = delimRegex;
    this.delimMode = delimMode;
  }

  /**
   * Must be synchronized to isolate watcher
   */
  @Override
  synchronized public void open() throws IOException {
    Preconditions.checkState(watcher == null,
        "Attempting to open an already open TailDirSource (" + dir + ", \""
            + regex + "\")");
    subdirWatcherMap = new ConcurrentHashMap<String, DirWatcher>();
    watcher = createWatcher(dir, regex, recurseDepth);
    dirChecked = true;
    watcher.start();
    tail.open();
  }

  private DirWatcher createWatcher(File dir, final String regex,
      final int recurseDepth) {
    // 250 ms between checks
    DirWatcher watcher = new DirWatcher(dir, new RegexFileFilter(regex), 250);
    watcher.addHandler(new DirChangeHandler() {
      Map<String, Cursor> curmap = new HashMap<String, Cursor>();

      @Override
      public void fileCreated(File f) {
        if (f.isDirectory()) {
          if (recurseDepth <= 0) {
            LOG.debug("Tail dir will not read or recurse "
                + "into subdirectory " + f + ", this watcher recurseDepth: "
                + recurseDepth);
            return;
          }

          LOG.info("added dir " + f + ", recurseDepth: " + (recurseDepth - 1));
          DirWatcher watcher = createWatcher(f, regex, recurseDepth - 1);
          watcher.start();
          subdirWatcherMap.put(f.getPath(), watcher);
          subdirsAdded.incrementAndGet();
          return;
        }

        // Add a new file to the multi tail.
        LOG.info("added file " + f);
        Cursor c;
        if (delimRegex == null) {
          if (startFromEnd && !dirChecked) {
            // init cursor positions on first dir check when startFromEnd is set
            // to true
            c = new Cursor(tail.sync, f, f.length(), f.length(), f
                .lastModified());
            try {
              c.initCursorPos();
            } catch (InterruptedException e) {
              LOG.error("Initializing of cursor failed", e);
              c.close();
              return;
            }
          } else {
            c = new Cursor(tail.sync, f);
          }
        } else {
          // special delimiter modes
          if (startFromEnd && !dirChecked) {
            // init cursor positions on first dir check when startFromEnd is set
            // to true
            c = new CustomDelimCursor(tail.sync, f, f.length(), f.length(), f
                .lastModified(), delimRegex, delimMode);
            try {
              c.initCursorPos();
            } catch (InterruptedException e) {
              LOG.error("Initializing of custom delimiter cursor failed", e);
              c.close();
              return;
            }
          } else {
            c = new CustomDelimCursor(tail.sync, f, delimRegex, delimMode);
          }
        }

        curmap.put(f.getPath(), c);
        tail.addCursor(c);
        filesAdded.incrementAndGet();
      }

      @Override
      public void fileDeleted(File f) {
        LOG.debug("handling deletion of file " + f);
        String fileName = f.getPath();
        // we cannot just check here with f.isDirectory() because f was deleted
        // and f.isDirectory() will return false always
        DirWatcher watcher = subdirWatcherMap.remove(fileName);
        if (watcher != null) {
          LOG.info("removed dir " + f);
          LOG.info("stopping watcher for dir: " + f);
          // stop is not thread-safe, but since this watcher belongs only to
          // this current thread it is safe to call it
          watcher.stop();
          // calling check explicitly to notify about deleted subdirs,
          // so that subdirs watchers can be stopped
          watcher.check();
          subdirsDeleted.incrementAndGet();
          return;
        }

        Cursor c = curmap.remove(fileName);
        // this check may seem unneeded but there are cases which it handles,
        // e.g. if unwatched subdirectory was removed c is null.
        if (c != null) {
          LOG.info("removed file " + f);
          tail.removeCursor(c);
          filesDeleted.incrementAndGet();
        }
      }

    });

    // Separate check is needed to init cursor positions
    // (to the end of the files in dir)
    if (startFromEnd) {
      watcher.check();
    }
    return watcher;
  }

  /**
   * Must be synchronized to isolate watcher
   * @throws InterruptedException
   */
  @Override
  synchronized public void close() throws IOException, InterruptedException {
    tail.close();
    this.watcher.stop();
    this.watcher = null;
    for (DirWatcher watcher : subdirWatcherMap.values()) {
      watcher.stop();
    }
    subdirWatcherMap = null;
  }

  @Override
  synchronized public ReportEvent getMetrics() {
    ReportEvent rpt = super.getMetrics();
    rpt.setLongMetric(A_FILESADDED, filesAdded.get());
    rpt.setLongMetric(A_FILESDELETED, filesDeleted.get());
    rpt.setLongMetric(A_SUBDIRSADDED, subdirsAdded.get());
    rpt.setLongMetric(A_SUBDIRSDELETED, subdirsDeleted.get());
    rpt.setLongMetric(A_FILESPRESENT, tail.cursors.size());
    return rpt;
  }

  @Override
  public Event next() throws IOException, InterruptedException {
    // this cannot be in synchronized because it has a
    // blocking call to a queue inside it.
    Event e = tail.next();

    synchronized (this) {
      updateEventProcessingStats(e);
      return e;
    }
  }

  public static SourceBuilder builder() {
    return new SourceBuilder() {
      @Override
      public EventSource build(Context ctx, String... argv) {
        Preconditions
            .checkArgument(argv.length >= 1 && argv.length <= 4, USAGE);

        String regex = ".*"; // default to accepting all
        if (argv.length >= 2) {
          regex = argv[1];
        }
        boolean startFromEnd = false;
        if (argv.length >= 3) {
          startFromEnd = Boolean.parseBoolean(argv[2]);
        }
        int recurseDepth = 0;
        if (argv.length >= 4) {
          recurseDepth = Integer.parseInt(argv[3]);
          Preconditions.checkArgument(recurseDepth >= 0,
              "\"recurseDepth\" should be >= 0, but was: " + recurseDepth
                  + ".\n" + USAGE);
        }

        // delim regex, delim mode
        Pair<String, DelimMode> mode = TailSource.extractDelimContext(ctx);
        if (mode != null) {
          return new TailDirSource(new File(argv[0]), regex, startFromEnd,
              recurseDepth, mode.getLeft(), mode.getRight());
        }
        return new TailDirSource(new File(argv[0]), regex, startFromEnd,
            recurseDepth);

      }
    };
  }
} public class TailSource extends EventSource.Base {
  private static final Logger LOG = LoggerFactory.getLogger(TailSource.class);
  public static final String A_TAILSRCFILE = "tailSrcFile";

  private static int thdCount = 0;
  private volatile boolean done = false;

  private final long sleepTime; // millis
  final List<Cursor> cursors = new ArrayList<Cursor>();
  private final List<Cursor> newCursors = new ArrayList<Cursor>();
  private final List<Cursor> rmCursors = new ArrayList<Cursor>();

  // We "queue" only allowing a single Event.
  final SynchronousQueue<Event> sync = new SynchronousQueue<Event>();
  private TailThread thd = null;

  /**
   * Constructor for backwards compatibility.
   */
  public TailSource(File f, long offset, long waitTime) {
    this(f, offset, waitTime, false);
  }

  /**
   * Specify the file, the starting offset (something >=0) and wait time between
   * checks in millis. If startFromEnd is set, begin reading the file at the
   * end, not the beginning.
   */
  public TailSource(File f, long offset, long waitTime, boolean startFromEnd) {
    Preconditions.checkArgument(offset >= 0 || startFromEnd,
        "offset needs to be >=0 or startFromEnd needs to be true");
    Preconditions.checkNotNull(f);
    Preconditions.checkArgument(waitTime > 0);
    this.sleepTime = waitTime;

    // add initial cursor.
    long fileLen = f.length();
    long readOffset = startFromEnd ? fileLen : offset;
    long modTime = f.lastModified();
    Cursor c = new Cursor(sync, f, readOffset, fileLen, modTime);
    addCursor(c);

  }

  /** Custom delimiter version **/
  public TailSource(File f, long offset, long waitTime, boolean startFromEnd,
      String regex, DelimMode dm) {
    Preconditions.checkArgument(f != null, "Null File is an illegal argument");
    Preconditions.checkArgument(waitTime > 0,
        "waitTime <=0 is an illegal argument");
    Preconditions.checkArgument(regex != null,
        "Null regex is an illegal argument");
    Preconditions.checkArgument(dm != null,
        "Null Delimiter mode is an illegal argument");
    this.sleepTime = waitTime;

    // add initial cursor.
    long fileLen = f.length();
    long readOffset = startFromEnd ? fileLen : offset;
    long modTime = f.lastModified();

    Cursor c = new CustomDelimCursor(sync, f, readOffset, fileLen, modTime,
        regex, dm);
    addCursor(c);
  }

  /**
   * This creates an empty tail source. It expects something else to add cursors
   * to it
   */
  public TailSource(long waitTime) {
    this.sleepTime = waitTime;
  }

  /**
   * This is the main driver thread that runs through the file cursor list
   * checking for updates and sleeping if there are none.
   */
  class TailThread extends Thread {

    TailThread() {
      super("TailThread-" + thdCount++);
    }

    @Override
    public void run() {
      try {
        // initialize based on initial settings.
        for (Cursor c : cursors) {
          c.initCursorPos();
        }

        while (!done) {
          synchronized (newCursors) {
            cursors.addAll(newCursors);
            newCursors.clear();
          }

          synchronized (rmCursors) {
            cursors.removeAll(rmCursors);
            for (Cursor c : rmCursors) {
              c.flush();
            }
            rmCursors.clear();
          }

          boolean madeProgress = false;
          for (Cursor c : cursors) {
            LOG.debug("Progress loop: " + c.file);
            if (c.tailBody()) {
              madeProgress = true;
            }
          }

          if (!madeProgress) {
            Clock.sleep(sleepTime);
          }
        }
        LOG.debug("Tail got done flag");
      } catch (InterruptedException e) {
        LOG.error("Tail thread nterrupted: " + e.getMessage(), e);
      } finally {
        LOG.info("TailThread has exited");
      }

    }
  }

  /**
   * Add another file Cursor to tail concurrently.
   */
  synchronized void addCursor(Cursor cursor) {
    Preconditions.checkArgument(cursor != null);

    if (thd == null) {
      cursors.add(cursor);
      LOG.debug("Unstarted Tail has added cursor: " + cursor.file.getName());

    } else {
      synchronized (newCursors) {
        newCursors.add(cursor);
      }
      LOG.debug("Tail added new cursor to new cursor list: "
          + cursor.file.getName());
    }

  }

  /**
   * Remove an existing cursor to tail.
   */
  synchronized public void removeCursor(Cursor cursor) {
    Preconditions.checkArgument(cursor != null);
    if (thd == null) {
      cursors.remove(cursor);
    } else {

      synchronized (rmCursors) {
        rmCursors.add(cursor);
      }
    }

  }

  @Override
  public void close() throws IOException, InterruptedException {
    synchronized (this) {
      done = true;
      if (thd == null) {
        LOG.warn("TailSource double closed");
        return;
      }
      while (thd.isAlive() ){
        thd.join(100L);
        thd.interrupt();
      }
      thd = null;
    }
  }

  /**
   * This function will block when the end of all the files it is trying to tail
   * is reached.
   */
  @Override
  public Event next() throws IOException, InterruptedException {
    try {
      while (!done) {
        // This blocks on the synchronized queue until a new event arrives.
        Event e = sync.poll(100, TimeUnit.MILLISECONDS);
        if (e == null)
          continue; // nothing there, retry.
        updateEventProcessingStats(e);
        return e;
      }
      return null; // closed
    } catch (InterruptedException e1) {
      LOG.warn("next unexpectedly interrupted :" + e1.getMessage(), e1);
      throw e1;
    }
  }

  @Override
  synchronized public void open() throws IOException {
    if (thd != null) {
      throw new IllegalStateException("Attempted to open tail source twice!");
    }
    thd = new TailThread();
    thd.start();
  }

  /**
   * This takes a context and extracts the delimiter regex and dilimiter mode.
   * If no mode is specified it defaults to EXCLUDE mode. If no regex is
   * specified, null is returned.
   */
  public static Pair<String, DelimMode> extractDelimContext(Context ctx) {
    String delimRegex = ctx.getValue("delim");
    if (delimRegex == null) {
      // don't have a regex, return null;
      return null;
    }

    // figure out mode, and delimiters
    String delimModeStr = ctx.getValue("delimMode");
    DelimMode delimMode = DelimMode.EXCLUDE; // default to exclude mode
    if (delimModeStr != null) {
      if ("exclude".equals(delimModeStr)) {
        delimMode = DelimMode.EXCLUDE;
      } else if ("prev".equals(delimModeStr)) {
        delimMode = DelimMode.INCLUDE_PREV;
      } else if ("next".equals(delimModeStr)) {
        delimMode = DelimMode.INCLUDE_NEXT;
      }
    }
    return new Pair<String, DelimMode>(delimRegex, delimMode);

  }

  public static SourceBuilder builder() {
    return new SourceBuilder() {

      @Override
      public EventSource build(Context ctx, String... argv) {
        if (argv.length != 1 && argv.length != 2) {
          throw new IllegalArgumentException(
              "usage: tail(filename, [startFromEnd] {, delim=\"regex\", delimMode=\"exclude|prev|next\"}) ");
        }
        boolean startFromEnd = false;
        if (argv.length == 2) {
          startFromEnd = Boolean.parseBoolean(argv[1]);
        }

        // delim regex, delim mode
        Pair<String, DelimMode> mode = extractDelimContext(ctx);
        if (mode == null) {
          // normal '\n' delimiter in exclude mode
          return new TailSource(new File(argv[0]), 0, FlumeConfiguration.get()
              .getTailPollPeriod(), startFromEnd);
        }

        return new TailSource(new File(argv[0]), 0, FlumeConfiguration.get()
            .getTailPollPeriod(), startFromEnd, mode.getLeft(), mode.getRight());
      }
    };
  }

  public static SourceBuilder multiTailBuilder() {
    return new SourceBuilder() {

      @Override
      public EventSource build(Context ctx, String... argv) {
        Preconditions.checkArgument(argv.length >= 1,
            "usage: multitail(file1[, file2[, ...]]) ");
        boolean startFromEnd = false;
        long pollPeriod = FlumeConfiguration.get().getTailPollPeriod();
        TailSource src = null;

        // delim regex, delim mode
        Pair<String, DelimMode> mode = extractDelimContext(ctx);

        for (int i = 0; i < argv.length; i++) {
          if (mode == null) {
            // default '\n' exclude mode
            if (src == null) {
              src = new TailSource(new File(argv[i]), 0, pollPeriod,
                  startFromEnd);
            } else {
              src.addCursor(new Cursor(src.sync, new File(argv[i])));
            }
          } else {
            // custom delimiters and delimiter modes
            if (src == null) {
              src = new TailSource(new File(argv[i]), 0, pollPeriod,
                  startFromEnd, mode.getLeft(), mode.getRight());
            } else {
              src.addCursor(new CustomDelimCursor(src.sync, new File(argv[i]),
                  mode.getLeft(), mode.getRight()));
            }

          }
        }
        return src;
      }
    };
  }

} public class PrioritizedThriftEventSource extends EventSource.Base {
  static int DEFAULT_QUEUE_SIZE = 1000;

  static final Logger LOG = LoggerFactory
      .getLogger(PrioritizedThriftEventSource.class);
  int port;
  ThriftFlumeEventServer svr;
  TSaneThreadPoolServer server;

  public static class EventQueue extends PriorityBlockingQueue<Event> {

    /**
     * (jon) don't know where this number comes from.
     */
    private static final long serialVersionUID = 7280524922090162382L;

    // We want to prioritize more important messages and older messages.
    public EventQueue(int size) {
      super(size, new Comparator<Event>() {
        @Override
        public int compare(Event o1, Event o2) {
          // higher priority entries are more important, and then older entries
          // are more important.
          int priorityDiff = o1.getPriority().ordinal()
              - o2.getPriority().ordinal();
          if (priorityDiff != 0)
            return priorityDiff;

          // smaller time means earlier and more important. Priority queue puts
          // takes the smallest first, which means we want a negative number if
          // o1 is older than o2.

          // Interesting -- we can get the same timestamp!
          long tdelta = (o1.getTimestamp() - o2.getTimestamp());
          if (tdelta != 0)
            return (int) tdelta;

          long ndelta = (o1.getNanos() - o2.getNanos());
          return (int) ndelta;
        }
      });
    }
  }

  final BlockingQueue<Event> q;

  /**
   * Creates a new prioritized event source on port port with event queue size
   * qsize.
   */
  public PrioritizedThriftEventSource(int port, int qsize) {
    this.port = port;
    this.svr = new ThriftFlumeEventServer();
    this.q = new EventQueue(qsize);
  }

  public PrioritizedThriftEventSource(int port) {
    this(port, DEFAULT_QUEUE_SIZE);
  }

  @Override
  public void open() throws IOException {

    try {
      ThriftFlumeEventServer.Processor processor = new ThriftFlumeEventServer.Processor(
          new ThriftFlumeEventServerImpl(new EventSink.Base() {
            @Override
            public void append(Event e) throws IOException,
                InterruptedException {
              q.add(e);
              super.append(e);
            }
          }, false));
      Factory protFactory = new TBinaryProtocol.Factory(true, true);

      TSaneServerSocket serverTransport = new TSaneServerSocket(port);
      server = new TSaneThreadPoolServer(processor, serverTransport,
          protFactory);
      LOG.info(String.format(
          "Starting blocking thread pool server on port %d...", port));

      server.start();

    } catch (TTransportException e) {
      e.printStackTrace();
      throw new IOException("Failed to create event server " + e);
    }
  }

  @Override
  public void close() throws IOException {
    if (server == null) {
      LOG.info(String.format("Server on port %d was already closed!", port));
      return;
    }
    server.stop();
    LOG.info(String.format("Closed server on port %d...", port));
  }

  @Override
  public Event next() throws IOException {

    try {
      Event e = q.take();
      updateEventProcessingStats(e);
      return e;
    } catch (InterruptedException e) {
      e.printStackTrace();
      throw new IOException("Waiting for queue element was interrupted! " + e);
    }
  }

  public static SourceBuilder builder() {
    return new SourceBuilder() {

      @Override
      public EventSource build(Context ctx, String... argv) {
        Preconditions.checkArgument(argv.length == 1, "usage: tsource(port)");

        int port = Integer.parseInt(argv[0]);

        return new PrioritizedThriftEventSource(port);
      }

    };
  }
}
import java.lang.reflect.Method;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.cloudera.flume.collector.CollectorSource;
import com.cloudera.flume.core.EventSource;
import com.cloudera.flume.core.PollingSource;
import com.cloudera.flume.core.EventSource.StubSource;
import com.cloudera.flume.handlers.avro.AvroEventSource;
import com.cloudera.flume.handlers.console.JLineStdinSource;
import com.cloudera.flume.handlers.debug.Log4jTextFileSource;
import com.cloudera.flume.handlers.debug.NoNlASCIISynthSource;
import com.cloudera.flume.handlers.debug.NoNlSynthSource;
import com.cloudera.flume.handlers.debug.NullSource;
import com.cloudera.flume.handlers.debug.StdinSource;
import com.cloudera.flume.handlers.debug.SynthSource;
import com.cloudera.flume.handlers.debug.SynthSourceRndSize;
import com.cloudera.flume.handlers.debug.TextFileSource;
import com.cloudera.flume.handlers.exec.ExecNioSource;
import com.cloudera.flume.handlers.hdfs.SeqfileEventSource;
import com.cloudera.flume.handlers.irc.IrcSource;
import com.cloudera.flume.handlers.rpc.RpcSource;
import com.cloudera.flume.handlers.scribe.ScribeEventSource;
import com.cloudera.flume.handlers.syslog.SyslogTcpSource;
import com.cloudera.flume.handlers.syslog.SyslogTcpSourceThreads;
import com.cloudera.flume.handlers.syslog.SyslogUdpSource;
import com.cloudera.flume.handlers.text.TailDirSource;
import com.cloudera.flume.handlers.text.TailSource;
import com.cloudera.flume.handlers.thrift.PrioritizedThriftEventSource;
import com.cloudera.flume.handlers.thrift.ThriftEventSource;
import com.cloudera.flume.handlers.twitter.TwitterStreamSource;
import com.cloudera.util.Pair;
/**
 * Licensed to Cloudera, Inc. under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  Cloudera, Inc. licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.cloudera.flume.conf;




/**
 * This factory creates event sources. It currently requires a recompile when
 * new sources are added.
 */
public class SourceFactoryImpl extends SourceFactory {
  static final Logger LOG = LoggerFactory.getLogger(SourceFactoryImpl.class);

  static Object[][] sourceList = {
      // high level sources
      { "logicalSource", StubSource.builder() },
      { "autoCollectorSource", StubSource.builder(0, 0) },// no args allowed
      { "collectorSource", CollectorSource.builder() },
      { "fail", StubSource.builder() },

      // low level Sources
      { "null", NullSource.builder() },
      { "stdin", StdinSource.builder() },
      { "console", JLineStdinSource.builder() },

      // creates AvroEventSource or ThriftEventSource
      { "rpcSource", RpcSource.builder() },
      { "thriftSource", ThriftEventSource.builder() },
      { "avroSource", AvroEventSource.builder() },
      { "text", TextFileSource.builder() },
      { "tail", TailSource.builder() },
      { "multitail", TailSource.multiTailBuilder() },
      { "tailDir", TailDirSource.builder() },
      { "seqfile", SeqfileEventSource.builder() },
      { "syslogUdp", SyslogUdpSource.builder() },
      { "syslogTcp", SyslogTcpSourceThreads.builder() },