@Test public void emitMultipleChangesInSubfolders() 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> folders = new ArrayList<>(); Path parentFolder = BASE_PATH; for (int i = 0; i < halfRequested; i++) { final Path newFolder = Files.createDirectory(parentFolder.resolve("testFolder" + i)); folders.add(newFolder); parentFolder = newFolder; } for (int i = 0; i < halfRequested; i++) { probe.expectNext(); } for (int i = halfRequested - 1; i >= 0; i--) { Files.delete(folders.get(i)); } for (int i = halfRequested - 1; i >= 0; 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);  }