@Test
public void testExtractLines() throws InterruptedException {

  LinkedBlockingQueue<Event> q = new LinkedBlockingQueue<Event>();
  long maxSz = FlumeConfiguration.get().getEventMaxSizeBytes();
  ByteBuffer buf = ByteBuffer.allocate((int) maxSz * 4);
  // Empty Case
  boolean ret = ExecNioSource.extractLines(buf, "cmd", "tag", q);
  assertFalse(ret);
  assertEquals(0, q.size());
  // Overflow Case
  // this should fill the buffer and but progress should be made.
  for (int i = 0; i < maxSz; i++) {
    buf.put((byte) 'a');
  }
  ret = ExecNioSource.extractLines(buf, "cmd", "tag", q);
  assertFalse(ret);
  assertEquals(maxSz, buf.position()); // should have buffer cleared
  assertEquals(0, q.size());
  buf.put((byte) '\n');
  ret = ExecNioSource.extractLines(buf, "cmd", "tag", q);
  assertTrue(ret);
  assertEquals(0, buf.position()); // should have buffer cleared
  assertEquals(1, q.size());
  Event e = q.poll();
  assertEquals(maxSz, e.getBody().length);
  // Normal Case
  for (int i = 0; i < maxSz; i++) {
    if (i % 1024 == 1023) {
      buf.put((byte) '\n');
    } else {
      buf.put((byte) 'b');
    }
  }
  ret = ExecNioSource.extractLines(buf, "cmd", "tag", q);
  assertTrue(ret);
  assertEquals(0, buf.position()); // should have buffer cleared
  assertEquals(32, q.size());
  for (int i = 0; i < 32; i++) {
    e = q.poll();
    assertEquals(1023, e.getBody().length);
  }
}