### Question:
FileProtocolClient implements ProtocolClient { @Override public Observable<Content> observeResource(Form form) { Path path = hrefToPath.apply(form.getHref()); Path directory = path.getParent(); FileSystem fileSystem = directory.getFileSystem(); return Observable.using( fileSystem::newWatchService, service -> Observable.<Content>create(source -> { directory.register(service, ENTRY_MODIFY, ENTRY_DELETE, ENTRY_CREATE); pollFileChange(path, service, source); source.onComplete(); }), WatchService::close ) .doOnSubscribe(d -> subscriptionsCount.incrementAndGet()) .doOnDispose(subscriptionsCount::decrementAndGet) .subscribeOn(Schedulers.io()); } FileProtocolClient();   FileProtocolClient(Function<String, Path> hrefToPath); @Override CompletableFuture<Content> readResource(Form form); @Override CompletableFuture<Content> writeResource(Form form, Content content); @Override Observable<Content> observeResource(Form form);  }### Answer:
@Test public void subscribeResourceShouldCreateWatchService() throws IOException { when(hrefToPath.apply(any())).thenReturn(path); when(path.getParent()).thenReturn(directory); when(directory.getFileSystem()).thenReturn(fileSystem); LambdaObserver<Content> observer = new LambdaObserver<>(n -> { }, e -> { }, () -> { }, s -> { }); FileProtocolClient client = new FileProtocolClient(hrefToPath); client.observeResource(form).subscribe(observer); verify(directory, timeout(5 * 1000L)).register(any(), any()); }
@Test public void subscribeResourceShouldCloseWatchServiceWhenObserverIsDone() throws IOException, InterruptedException { when(hrefToPath.apply(any())).thenReturn(path); when(path.getParent()).thenReturn(directory); when(directory.getFileSystem()).thenReturn(fileSystem); when(fileSystem.newWatchService()).thenReturn(watchService); when(watchService.take()).thenAnswer(new AnswersWithDelay(5 * 1000L, new Returns(null))); FileProtocolClient client = new FileProtocolClient(hrefToPath); Disposable subscribe = client.observeResource(form).subscribe(); verify(directory, timeout(5 * 1000L)).register(any(), any()); subscribe.dispose(); verify(watchService, timeout(5 * 1000L)).close(); }