@Test public void failStageWhenBufferFull() throws Exception { final TestSubscriber.Probe<Pair<Path, DirectoryChange>> probe = TestSubscriber.probe(system); final int maxBufferSize = 2; final int numberOfChanges = 3; RecursiveDirectoryChangesSource.create(BASE_PATH, Duration.of(250, ChronoUnit.MILLIS), maxBufferSize) .runWith(Sink.fromSubscriber(probe), materializer); probe.request(numberOfChanges); final List<Path> files = new ArrayList<>(); for (int i = 0; i < numberOfChanges; i++) { final Path file = Files.createFile(BASE_PATH.resolve("test2files" + i)); files.add(file); } Throwable throwable = probe.expectError(); assertTrue("Not a runtime exception", throwable instanceof RuntimeException); if(Files.exists(BASE_PATH)){ Files.walk(BASE_PATH) .sorted(Comparator.reverseOrder()) .map(Path::toFile) .forEach(File::delete); } probe.cancel(); }
public static Source<Pair<Path, DirectoryChange>, NotUsed> create(Path directoryPath, Duration pollInterval, int maxBufferSize) { return Source.fromGraph(new DirectoryWatcherSourceStage(directoryPath, pollInterval, maxBufferSize)); }
RecursiveDirectoryChangesSource { public static Source<Pair<Path, DirectoryChange>, NotUsed> create(Path directoryPath, Duration pollInterval, int maxBufferSize) { return Source.fromGraph(new DirectoryWatcherSourceStage(directoryPath, pollInterval, maxBufferSize)); } }
RecursiveDirectoryChangesSource { public static Source<Pair<Path, DirectoryChange>, NotUsed> create(Path directoryPath, Duration pollInterval, int maxBufferSize) { return Source.fromGraph(new DirectoryWatcherSourceStage(directoryPath, pollInterval, maxBufferSize)); }  }
RecursiveDirectoryChangesSource { public static Source<Pair<Path, DirectoryChange>, NotUsed> create(Path directoryPath, Duration pollInterval, int maxBufferSize) { return Source.fromGraph(new DirectoryWatcherSourceStage(directoryPath, pollInterval, maxBufferSize)); }  static Source<Pair<Path, DirectoryChange>, NotUsed> create(Path directoryPath, Duration pollInterval, int maxBufferSize); }
RecursiveDirectoryChangesSource { public static Source<Pair<Path, DirectoryChange>, NotUsed> create(Path directoryPath, Duration pollInterval, int maxBufferSize) { return Source.fromGraph(new DirectoryWatcherSourceStage(directoryPath, pollInterval, maxBufferSize)); }  static Source<Pair<Path, DirectoryChange>, NotUsed> create(Path directoryPath, Duration pollInterval, int maxBufferSize);  }