### Question:
CoapProtocolClient implements ProtocolClient { @Override public Observable<Content> observeResource(Form form) { String url = form.getHref(); return Observable.using( () -> clientCreator.apply(url), client -> Observable.create(source -> { log.debug("CoapClient subscribe to '{}'", url); client.observe(new CoapHandler() { @Override public void onLoad(CoapResponse response) { String type = MediaTypeRegistry.toString(response.getOptions().getContentFormat()); byte[] body = response.getPayload(); Content output = new Content(type, body); if (response.isSuccess()) { log.debug("Next data received for subscription '{}'", url); source.onNext(output); } else { try { String error = ContentManager.contentToValue(output, new StringSchema()); source.onError(new ProtocolClientException(error)); log.debug("Error received for subscription '{}': {}", url, error); } catch (ContentCodecException e) { source.onError(new ProtocolClientException(e)); log.debug("Error received for subscription '{}': {}", url, e.getMessage()); } } } @Override public void onError() { source.onError(new ProtocolClientException("Error received for subscription '" + url + "'")); log.debug("Error received for subscription '{}'", url); } }); }), CoapClient::shutdown ); } CoapProtocolClient(ExecutorService executor);   CoapProtocolClient(Function<String, CoapClient> 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);  }### Answer:
@Test public void observeResourceShouldCreateCoapRequest() { when(form.getHref()).thenReturn("coap: when(clientCreator.apply(any())).thenReturn(coapClient); CoapProtocolClient client = new CoapProtocolClient(clientCreator); client.observeResource(form).subscribe(); verify(coapClient, timeout(1 * 1000L)).observe(any()); }