@Test public void sourceShouldEmitOnSubdirectoryChanges() throws IOException { 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 createdSubfolder = Files.createDirectory(BASE_PATH.resolve("subfolder1")); final Pair<Path, DirectoryChange> subDirCreatedPair = probe.expectNext(); assertEquals(DirectoryChange.Creation, subDirCreatedPair.second()); assertEquals(createdSubfolder, subDirCreatedPair.first()); final Path createdFile = Files.createFile(createdSubfolder.resolve("test1file1.sample")); final Pair<Path, DirectoryChange> fileCreatedPair = probe.requestNext(); assertEquals(DirectoryChange.Creation, fileCreatedPair.second()); assertEquals(createdFile, fileCreatedPair.first()); Files.delete(createdFile); final Pair<Path, DirectoryChange> fileDeletedPair = probe.requestNext(); assertEquals(DirectoryChange.Deletion, fileDeletedPair.second()); assertEquals(createdFile, fileDeletedPair.first()); Files.delete(createdSubfolder); Pair<Path, DirectoryChange> folderDeletedPair = probe.requestNext(); if (folderDeletedPair.second().equals(DirectoryChange.Modification)){ folderDeletedPair = probe.requestNext(); } assertEquals(DirectoryChange.Deletion, folderDeletedPair.second()); assertEquals(createdSubfolder, folderDeletedPair.first()); 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);  }