@Test public void testConsumeFileRandomlyNewFile() throws Exception { if (SystemUtils.IS_OS_WINDOWS) { return; } final ReliableEventReader reader = new ReliableSpoolingFileEventReader.Builder().spoolDirectory(WORK_DIR) .consumeOrder(ConsumeOrder.RANDOM) .sourceCounter(new SourceCounter("test")) .build(); File fileName = new File(WORK_DIR, "new-file"); FileUtils.write(fileName, "New file created in the end. Shoud be read randomly.\n"); Set<String> expected = Sets.newHashSet(); int totalFiles = WORK_DIR.listFiles().length; final Set<String> actual = Sets.newHashSet(); ExecutorService executor = Executors.newSingleThreadExecutor(); final Semaphore semaphore1 = new Semaphore(0); final Semaphore semaphore2 = new Semaphore(0); Future<Void> wait = executor.submit(new Callable<Void>() { @Override public Void call() throws Exception { readEventsForFilesInDir(WORK_DIR, reader, actual, semaphore1, semaphore2); return null; } }); semaphore1.acquire(); File finalFile = new File(WORK_DIR, "t-file"); FileUtils.write(finalFile, "Last file"); semaphore2.release(); wait.get(); int listFilesCount = ((ReliableSpoolingFileEventReader)reader).getListFilesCount(); finalFile.delete(); createExpectedFromFilesInSetup(expected); expected.add(""); expected.add("New file created in the end. Shoud be read randomly."); expected.add("Last file"); Assert.assertTrue(listFilesCount < (totalFiles + 2)); Assert.assertEquals(expected, actual); }
@VisibleForTesting int getListFilesCount() { return listFilesCount; }
ReliableSpoolingFileEventReader implements ReliableEventReader { @VisibleForTesting int getListFilesCount() { return listFilesCount; } }
ReliableSpoolingFileEventReader implements ReliableEventReader { @VisibleForTesting int getListFilesCount() { return listFilesCount; } private  ReliableSpoolingFileEventReader(File spoolDirectory,
      String completedSuffix, String includePattern, String ignorePattern, String trackerDirPath,
      boolean annotateFileName, String fileNameHeader,
      boolean annotateBaseName, String baseNameHeader,
      String deserializerType, Context deserializerContext,
      String deletePolicy, String trackingPolicy, String inputCharset,
      DecodeErrorPolicy decodeErrorPolicy,
      ConsumeOrder consumeOrder, boolean recursiveDirectorySearch,
                                          SourceCounter sourceCounter); }
ReliableSpoolingFileEventReader implements ReliableEventReader { @VisibleForTesting int getListFilesCount() { return listFilesCount; } private  ReliableSpoolingFileEventReader(File spoolDirectory,
      String completedSuffix, String includePattern, String ignorePattern, String trackerDirPath,
      boolean annotateFileName, String fileNameHeader,
      boolean annotateBaseName, String baseNameHeader,
      String deserializerType, Context deserializerContext,
      String deletePolicy, String trackingPolicy, String inputCharset,
      DecodeErrorPolicy decodeErrorPolicy,
      ConsumeOrder consumeOrder, boolean recursiveDirectorySearch,
                                          SourceCounter sourceCounter); String getLastFileRead(); Event readEvent(); List<Event> readEvents(int numEvents); @Override void close(); @Override void commit(); }
ReliableSpoolingFileEventReader implements ReliableEventReader { @VisibleForTesting int getListFilesCount() { return listFilesCount; } private  ReliableSpoolingFileEventReader(File spoolDirectory,
      String completedSuffix, String includePattern, String ignorePattern, String trackerDirPath,
      boolean annotateFileName, String fileNameHeader,
      boolean annotateBaseName, String baseNameHeader,
      String deserializerType, Context deserializerContext,
      String deletePolicy, String trackingPolicy, String inputCharset,
      DecodeErrorPolicy decodeErrorPolicy,
      ConsumeOrder consumeOrder, boolean recursiveDirectorySearch,
                                          SourceCounter sourceCounter); String getLastFileRead(); Event readEvent(); List<Event> readEvents(int numEvents); @Override void close(); @Override void commit();  }