@Test public void emitMultipleChanges() throws Exception { final TestSubscriber.Probe<Pair<Path, DirectoryChange>> probe = TestSubscriber.probe(system); final int numberOfChanges = 10; RecursiveDirectoryChangesSource.create(BASE_PATH, Duration.of(250,ChronoUnit.MILLIS), 200) .runWith(Sink.fromSubscriber(probe), materializer); probe.request(numberOfChanges); final int halfRequested = numberOfChanges / 2; final List<Path> files = new ArrayList<>(); for (int i = 0; i < halfRequested; i++) { final Path file = Files.createFile(BASE_PATH.resolve("test2files" + i)); files.add(file); } for (int i = 0; i < halfRequested; i++) { probe.expectNext(); } for (int i = 0; i < halfRequested; i++) { Files.delete(files.get(i)); } for (int i = 0; i < halfRequested; i++) { probe.expectNext(); } Files.delete(BASE_PATH); 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);  }