@Test
public void testTailSource() throws IOException, FlumeSpecException,
    InterruptedException {

  File f = FileUtil.createTempFile("temp", ".tmp");
  f.deleteOnExit();
  final CompositeSink snk = new CompositeSink(new ReportTestingContext(),
      "{ delay(50) => counter(\"count\") }");
  final EventSource src = TailSource.builder().build(
      LogicalNodeContext.testingContext(), f.getAbsolutePath());
  final CountDownLatch done = new CountDownLatch(1);
  final int count = 30;
  runDriver(src, snk, done, count);
  FileWriter fw = new FileWriter(f);
  for (int i = 0; i < count; i++) {
    fw.append("Line " + i + "\n");
    fw.flush();
  }
  fw.close();
  assertTrue(done.await(10, TimeUnit.SECONDS));
  CounterSink ctr = (CounterSink) ReportManager.get().getReportable("count");
  assertEquals(count, ctr.getCount());
}