### Question:
ReactiveRpcStreamerWrapper implements ReactiveRpcStreamer { @Override public CompletableFuture<ResponseMessage> response(RequestMessage.Builder requestMessage) { if (executor == null) { return CompletableFuture.supplyAsync(responseSupplier(requestMessage)); } else { return CompletableFuture.supplyAsync(responseSupplier(requestMessage), executor); } } ReactiveRpcStreamerWrapper(RpcStreamer rpcStreamer);  ReactiveRpcStreamerWrapper(RpcStreamer rpcStreamer, Executor executor); @Override void attach(RpcConnection rpcConnection); @Override CompletableFuture<ResponseMessage> response(RequestMessage.Builder requestMessage); @Override Flow.Publisher<RequestMessage> requestsFlow(); @Override Flow.Publisher<NotificationMessage> notificationsFlow();  }### Answer:
@Test public void testSendRequestCompletable() throws IOException { var message = new RequestMessage.Builder("test"); var preparedResponse = new ResponseMessage.Builder("test").withId(25).build(); doAnswer(invocationOnMock -> { RpcListener.ResponseCallback responseCallback = (RpcListener.ResponseCallback) invocationOnMock.getArguments()[1]; responseCallback.responseReceived(25, preparedResponse); return null; }).when(rpcStreamer).send(eq(message), any()); var countDownLatch = new CountDownLatch(1); reactiveRpcStreamerWrapper.response(message).thenAccept(responseMessage -> { try { assertTrue(countDownLatch.await(100, TimeUnit.MILLISECONDS)); } catch (InterruptedException e) { e.printStackTrace(); } }); countDownLatch.countDown(); var builderArgumentCaptor = ArgumentCaptor.forClass(RequestMessage.Builder.class); verify(rpcStreamer, timeout(100)).send(builderArgumentCaptor.capture(), any()); assertEquals(message, builderArgumentCaptor.getValue()); }