source: class TestTailDirSource {
  
    @Test
    public void testTailDirsRecursively() throws IOException, FlumeSpecException,
        InterruptedException {
  
target:     File tmpdir = FileUtil.mktempdir();
      // generating files: emulating their existence prior to sink opening
      genFiles(tmpdir, "2tail-old", 10, 100);
  
      // creating subdirs with data, i.e. structure is:
      // root
      // `-- subdir-level1
      // `-- subdir-level2
      // `-- subdir-level3
      // NOTE: the deepest subdir-level3 shouldn't be watched as recurse-depth is
      // 2.
      File subDirL1 = new File(tmpdir, "subdir-level1");
      subDirL1.mkdirs();
      genFiles(subDirL1, "2tail-old1", 10, 10);
      File subDirL2 = new File(subDirL1, "subdir-level2");
      subDirL2.mkdirs();
      genFiles(subDirL2, "2tail-old2", 10, 20);
      File subDirL3 = new File(subDirL2, "subdir-level3");
      subDirL3.mkdirs();
      genFiles(subDirL3, "2tail-old3", 10, 30);
  
      TailDirSource src = new TailDirSource(tmpdir, ".(2tail)?.*", true, 2);
      AccumulatorSink cnt = new AccumulatorSink("tailcount");
      DirectDriver drv = new DirectDriver(src, cnt);
  
      drv.start();
      Clock.sleep(1000);
      assertEquals(0, cnt.getCount());
  
      // adding lines to existing files
      addLinesToExistingFiles(tmpdir, 10);
      addLinesToExistingFiles(subDirL1, 20);
      addLinesToExistingFiles(subDirL2, 30);
      addLinesToExistingFiles(subDirL3, 40);
  
      Clock.sleep(1000);
      int expEventsCount = 10 * 10 + 10 * 20 + 10 * 30;
      assertEquals(expEventsCount, cnt.getCount());
  
      // generating new files
      genFiles(tmpdir, "2tail-new", 10, 40);
      genFiles(subDirL1, "2tail-new1", 10, 30);
      genFiles(subDirL2, "2tail-new2", 10, 20);
      genFiles(subDirL3, "2tail-new3", 10, 10);
  
      Clock.sleep(1000);
      expEventsCount += 10 * 40 + 10 * 30 + 10 * 20;
      assertEquals(expEventsCount, cnt.getCount());
  
      // creating new subdir on level 2
      File newSubDirL2 = new File(subDirL1, "subdir-level2-new");
      newSubDirL2.mkdirs();
      genFiles(newSubDirL2, "2tail-new2-new", 10, 100);
  
      Clock.sleep(1000);
      expEventsCount += 10 * 100;
      assertEquals(expEventsCount, cnt.getCount());
  
      // deleting some dirs
      FileUtil.rmr(subDirL3);
      FileUtil.rmr(subDirL2);
  
      Clock.sleep(1000);
      assertEquals(expEventsCount, cnt.getCount());
  
      // adding back deleted dir, checking that it is caught up
      subDirL2.mkdirs();
      genFiles(subDirL2, "2tail-new2", 10, 100);
      Clock.sleep(1000);
      expEventsCount += 10 * 100;
      assertEquals(expEventsCount, cnt.getCount());
  
      drv.stop();
      FileUtil.rmr(tmpdir);
  
      ReportEvent report = src.getMetrics();
      assertEquals(Long.valueOf(80),
          report.getLongMetric(TailDirSource.A_FILESADDED));
      assertEquals(Long.valueOf(20),
          report.getLongMetric(TailDirSource.A_FILESDELETED));
      assertEquals(Long.valueOf(4),
          report.getLongMetric(TailDirSource.A_SUBDIRSADDED));
      assertEquals(Long.valueOf(1),
          report.getLongMetric(TailDirSource.A_SUBDIRSDELETED));
    }
  
  }