RpcClient extends ch.squaredesk.nova.comm.rpc.RpcClient<String, RequestMessageMetaData, ReplyMessageMetaData> implements HttpClientInstanceListener { public <T> Single<RpcReply<InputStream>> sendRequestAndRetrieveResponseAsStream( T request, RequestMessageMetaData requestMessageMetaData, Function<T, String> requestTranscriber, Duration timeout) { return doRequest( request, requestMessageMetaData, requestTranscriber, Response::getResponseBodyAsStream, timeout ); } protected  RpcClient(String identifier,
                        AsyncHttpClient client,
                        Metrics metrics); @Override Single<RpcReply<U>> sendRequest(
            T request,
            RequestMessageMetaData requestMessageMetaData,
            Function<T, String> requestTranscriber,
            Function<String, U> replyTranscriber,
            Duration timeout); Single<RpcReply<InputStream>> sendRequestAndRetrieveResponseAsStream(
            T request,
            RequestMessageMetaData requestMessageMetaData,
            Function<T, String> requestTranscriber,
            Duration timeout); Map<String, String> getStandardHeadersForAllRequests(); void setStandardHeadersForAllRequests(Map<String, String> standardHeadersForAllRequests); @Override void httpClientInstanceCreated(AsyncHttpClient httpClient);  }
@Test void replyStreamCanProperlyBeRetrieved() throws Exception { Pair<Integer, SimpleHttpServer> portServerPair = createHttpServer(httpExchange -> { String response = "This is the response"; try { try { httpExchange.sendResponseHeaders(200, response.length()); OutputStream os = httpExchange.getResponseBody(); os.write(response.getBytes()); os.close(); } catch (Exception e) { throw new RuntimeException(e); } } catch (Exception e) { throw new RuntimeException(e); } }).blockingGet(); try { RequestInfo ri = new RequestInfo(HttpRequestMethod.GET); RequestMessageMetaData meta = new RequestMessageMetaData(new URL("http: Single<RpcReply<InputStream>> replySingle = sut.sendRequestAndRetrieveResponseAsStream("", meta, s -> s, Duration.ofSeconds(5)); TestObserver<RpcReply<InputStream>> observer = replySingle.test(); await().atMost(3, SECONDS).until(observer::valueCount, is(1)); observer.assertComplete(); RpcReply<InputStream> rpcReply = observer.values().get(0); BufferedReader replyStreamReader = new BufferedReader(new InputStreamReader(rpcReply.result)); assertThat(replyStreamReader.readLine(), is("This is the response")); } finally { portServerPair._2.close(); } }