### Question:
HttpProtocolClient implements ProtocolClient { @Override public Observable<Content> observeResource(Form form) { RequestConfig config = RequestConfig.custom() .setConnectTimeout((int) LONG_POLLING_TIMEOUT.toMillis()) .setConnectionRequestTimeout((int) LONG_POLLING_TIMEOUT.toMillis()) .setSocketTimeout((int) LONG_POLLING_TIMEOUT.toMillis()).build(); HttpUriRequest request = generateRequest(form); return Observable.using( () -> clientCreator.apply(config), client -> Observable.<Content>create(source -> { while (!source.isDisposed()) { log.debug("Sending '{}' to '{}'", request.getMethod(), request.getURI()); try { HttpResponse response = client.execute(request); Content content = checkResponse(response); log.debug("Next data received for Event connection"); source.onNext(content); } catch (IOException | ProtocolClientException e) { log.warn("Error received for Event connection", e); source.onError(e); } } }), Closeable::close ).subscribeOn(Schedulers.io()); } HttpProtocolClient();   HttpProtocolClient(HttpClient requestClient,
                       Function<RequestConfig, CloseableHttpClient> clientCreator); @Override CompletableFuture<Content> readResource(Form form); @Override CompletableFuture<Content> writeResource(Form form, Content content); @Override CompletableFuture<Content> invokeResource(Form form, Content content); @Override Observable<Content> observeResource(Form form); @Override boolean setSecurity(List<SecurityScheme> metadata, Object credentials);  }### Answer:
@Test public void subscribeResourceShouldCreateHttpRequest() throws IOException { when(clientCreator.apply(any())).thenReturn(httpClient); LambdaObserver<Content> observer = new LambdaObserver<>(n -> { }, e -> { }, () -> { }, s -> { }); client = new HttpProtocolClient(requestClient, clientCreator); client.observeResource(form).subscribe(observer); verify(httpClient, timeout(1 * 1000L)).execute(any()); }
@Test public void observeResourceShouldCloseHttpRequestWhenObserverIsDone() throws IOException { when(clientCreator.apply(any())).thenReturn(httpClient); doAnswer(new AnswersWithDelay(10 * 1000L, invocation -> httpResponse)).when(httpClient).execute(any()); client = new HttpProtocolClient(requestClient, clientCreator); Disposable subscribe = client.observeResource(form).subscribe(); verify(httpClient, timeout(1 * 1000L)).execute(any()); subscribe.dispose(); verify(httpClient, timeout(5 * 1000L)).close(); }