@Test public void sourceShouldEmitOnDirectoryChangesModifyFile() throws Exception { final TestSubscriber.Probe<Pair<Path, DirectoryChange>> probe = TestSubscriber.probe(system); RecursiveDirectoryChangesSource.create(BASE_PATH, Duration.of(250,ChronoUnit.MILLIS), 200) .runWith(Sink.fromSubscriber(probe), materializer); probe.request(1); final Path createdFile = Files.createFile(BASE_PATH.resolve("test1file1.sample")); final Pair<Path, DirectoryChange> pair1 = probe.expectNext(); assertEquals(DirectoryChange.Creation, pair1.second()); assertEquals(createdFile, pair1.first()); Files.write(createdFile, "Some data".getBytes()); final Pair<Path, DirectoryChange> pair2 = probe.requestNext(); assertEquals(DirectoryChange.Modification, pair2.second()); assertEquals(createdFile, pair2.first()); probe.cancel(); Files.delete(createdFile); Files.delete(BASE_PATH); }
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);  }