RpcClient extends ch.squaredesk.nova.comm.rpc.RpcClient<String, RequestMessageMetaData, ReplyMessageMetaData> implements HttpClientInstanceListener { @Override public <T, U> Single<RpcReply<U>> sendRequest( T request, RequestMessageMetaData requestMessageMetaData, Function<T, String> requestTranscriber, Function<String, U> replyTranscriber, Duration timeout) { return doRequest( request, requestMessageMetaData, requestTranscriber, response -> { String responseBody = response.getResponseBody(); return responseBody == null || responseBody.trim().isEmpty() ? null : replyTranscriber.apply(responseBody); }, 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 replyCanProperlyBeRetrieved() 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<String>> replySingle = sut.sendRequest("", meta, s -> s, s -> s, Duration.ofSeconds(5)); TestObserver<RpcReply<String>> observer = replySingle.test(); await().atMost(3, SECONDS).until(observer::valueCount, is(1)); observer.assertComplete(); RpcReply<String> rpcReply = observer.values().get(0); assertThat(rpcReply.result, is("This is the response")); } finally { portServerPair._2.close(); } }
@Test void replyHeadersCanProperlyBeRetrieved() throws Exception { Pair<Integer, SimpleHttpServer> portServerPair = createHttpServer(httpExchange -> { String response = "This is the response"; try { httpExchange.getResponseHeaders().add("header1", "value1"); httpExchange.getResponseHeaders().add("header2", "value2"); 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<String>> replySingle = sut.sendRequest("", meta, s -> s, s -> s, Duration.ofSeconds(5)); TestObserver<RpcReply<String>> observer = replySingle.test(); await().atMost(3, SECONDS).until(observer::valueCount, is(1)); observer.assertComplete(); RpcReply<String> rpcReply = observer.values().get(0); assertThat(rpcReply.metaData.details.headerParams.get("header1"), is("value1")); assertThat(rpcReply.metaData.details.headerParams.get("header2"), is("value2")); } finally { portServerPair._2.close(); } }
@Test void timeoutIsProperlyAppliedWhenDoingRpc() throws Exception { Pair<Integer, SimpleHttpServer> portServerPair = createHttpServer(httpExchange -> { }) .subscribeOn(Schedulers.newThread()) .blockingGet(); try { RequestInfo ri = new RequestInfo(HttpRequestMethod.GET); RequestMessageMetaData meta = new RequestMessageMetaData(new URL("http: Single<RpcReply<String>> replySingle = sut.sendRequest("", meta, s -> s, s -> s, Duration.ofSeconds(5)) .subscribeOn(Schedulers.io()) ; TestObserver<RpcReply<String>> observer = replySingle.test(); Thread.sleep(1000); observer.assertNotComplete(); observer.assertValueCount(0); Thread.sleep(2000); observer.assertNotComplete(); observer.assertValueCount(0); Thread.sleep(3000); observer.assertValueCount(0); observer.assertError(error -> error instanceof TimeoutException); } finally { portServerPair._2.close(); } }