HttpAdapter extends CommAdapter<String> implements HttpServerInstanceListener, HttpClientInstanceListener { public <U> Single<RpcReply<U>> sendGetRequest(String destination, Class<U> replyType) { return sendRequest(destination, null, new RequestInfo(HttpRequestMethod.GET), replyType, null ); } private  HttpAdapter(Builder builder); boolean isServerStarted(); FrozenHttpAdapter<T> freeze(Class<T> typeClass); void setStandardHeadersForAllRequests(Map<String, String> standardHeaders); Optional<Map<String, String>> getStandardHeadersForAllRequests(); Single<RpcReply<U>> sendGetRequest(String destination, Class<U> replyType); Single<RpcReply<U>> sendGetRequest(String destination, Map<String, String> headers, Class<U> replyType); Single<RpcReply<U>> sendGetRequest(
                String destination,
                Class<U> replyType,
                Duration timeout); Single<RpcReply<U>> sendGetRequest(
                String destination,
                Map<String, String> headers,
                Class<U> replyType,
                Duration timeout); Single<RpcReply<U>> sendGetRequest(String destination, Function<String, U> replyTranscriber); Single<RpcReply<U>> sendGetRequest(String destination, Map<String, String> headers, Function<String, U> replyTranscriber); Single<RpcReply<U>> sendGetRequest(
                String destination,
                Function<String, U> replyTranscriber,
                Duration timeout); Single<RpcReply<U>> sendGetRequest(
                String destination,
                Map<String, String> headers,
                Function<String, U> replyTranscriber,
                Duration timeout); Single<RpcReply<InputStream>> sendGetRequestAndRetrieveResponseAsStream(String destination); Single<RpcReply<InputStream>> sendGetRequestAndRetrieveResponseAsStream(String destination, Map<String, String> headers); Single<RpcReply<InputStream>> sendGetRequestAndRetrieveResponseAsStream(
            String destination,
            Duration timeout); Single<RpcReply<InputStream>> sendGetRequestAndRetrieveResponseAsStream(
            String destination,
            Map<String, String> headers,
            Duration timeout); Single<RpcReply<U>> sendPostRequest(
                String destination,
                T request,
                Class<U> replyType); Single<RpcReply<U>> sendPostRequest(
                String destination,
                T request,
                Map<String, String> headers,
                Class<U> replyType); Single<RpcReply<U>> sendPostRequest(
                String destination,
                T request,
                Class<U> replyType,
                Duration timeout); Single<RpcReply<U>> sendPostRequest(
                String destination,
                T request,
                Map<String, String> headers,
                Class<U> replyType,
                Duration timeout); Single<RpcReply<U>> sendPostRequest(
                String destination,
                T request,
                Function<String, U> replyTranscriber); Single<RpcReply<U>> sendPostRequest(
                String destination,
                T request,
                Map<String, String> headers,
                Function<String, U> replyTranscriber); Single<RpcReply<U>> sendPostRequest(
                String destination,
                T request,
                Function<String, U> replyTranscriber,
                Duration timeout); Single<RpcReply<U>> sendPostRequest(
                String destination,
                T request,
                Map<String, String> headers,
                Function<String, U> replyTranscriber,
                Duration timeout); Single<RpcReply<InputStream>> sendPostRequestAndRetrieveResponseAsStream(
            String destination,
            T request); Single<RpcReply<InputStream>> sendPostRequestAndRetrieveResponseAsStream(
            String destination,
            T request,
            Map<String, String> headers); Single<RpcReply<InputStream>> sendPostRequestAndRetrieveResponseAsStream(
            String destination,
            T request,
            Duration timeout); Single<RpcReply<InputStream>> sendPostRequestAndRetrieveResponseAsStream(
            String destination,
            T request,
            Map<String, String> headers,
            Duration timeout); Single<RpcReply<U>> sendPutRequest(
                String destination,
                T request,
                Class<U> replyType); Single<RpcReply<U>> sendPutRequest(
                String destination,
                T request,
                Map<String, String> headers,
                Class<U> replyType); Single<RpcReply<U>> sendPutRequest(
                String destination,
                T request,
                Class<U> replyType,
                Duration timeout); Single<RpcReply<U>> sendPutRequest(
                String destination,
                T request,
                Map<String, String> headers,
                Class<U> replyType,
                Duration timeout); Single<RpcReply<U>> sendPutRequest(
                String destination,
                T request,
                Function<String, U> replyTranscriber); Single<RpcReply<U>> sendPutRequest(
                String destination,
                T request,
                Map<String, String> headers,
                Function<String, U> replyTranscriber); Single<RpcReply<U>> sendPutRequest(
                String destination,
                T request,
                Function<String, U> replyTranscriber,
                Duration timeout); Single<RpcReply<U>> sendPutRequest(
                String destination,
                T request,
                Map<String, String> headers,
                Function<String, U> replyTranscriber,
                Duration timeout); Single<RpcReply<InputStream>> sendPutRequestAndRetrieveResponseAsStream(
                String destination,
                T request); Single<RpcReply<InputStream>> sendPutRequestAndRetrieveResponseAsStream(
                String destination,
                T request,
                Map<String, String> headers); Single<RpcReply<InputStream>> sendPutRequestAndRetrieveResponseAsStream(
                String destination,
                T request,
                Duration timeout); Single<RpcReply<InputStream>> sendPutRequestAndRetrieveResponseAsStream(
                String destination,
                T request,
                Map<String, String> headers,
                Duration timeout); Single<RpcReply<U>> sendPatchRequest(
                String destination,
                T request,
                Class<U> replyType); Single<RpcReply<U>> sendPatchRequest(
                String destination,
                T request,
                Map<String, String> headers,
                Class<U> replyType); Single<RpcReply<U>> sendPatchRequest(
                String destination,
                T request,
                Class<U> replyType,
                Duration timeout); Single<RpcReply<U>> sendPatchRequest(
                String destination,
                T request,
                Map<String, String> headers,
                Class<U> replyType,
                Duration timeout); Single<RpcReply<U>> sendPatchRequest(
                String destination,
                T request,
                Function<String, U> replyTranscriber); Single<RpcReply<U>> sendPatchRequest(
                String destination,
                T request,
                Map<String, String> headers,
                Function<String, U> replyTranscriber); Single<RpcReply<U>> sendPatchRequest(
                String destination,
                T request,
                Function<String, U> replyTranscriber,
                Duration timeout); Single<RpcReply<U>> sendPatchRequest(
                String destination,
                T request,
                Map<String, String> headers,
                Function<String, U> replyTranscriber,
                Duration timeout); Single<RpcReply<InputStream>> sendPatchRequestAndRetrieveResponseAsStream(
                String destination,
                T request); Single<RpcReply<InputStream>> sendPatchRequestAndRetrieveResponseAsStream(
                String destination,
                T request,
                Map<String, String> headers); Single<RpcReply<InputStream>> sendPatchRequestAndRetrieveResponseAsStream(
                String destination,
                T request,
                Duration timeout); Single<RpcReply<InputStream>> sendPatchRequestAndRetrieveResponseAsStream(
                String destination,
                T request,
                Map<String, String> headers,
                Duration timeout); Single<RpcReply<U>> sendDeleteRequest(
                String destination,
                T request,
                Class<U> replyType); Single<RpcReply<U>> sendDeleteRequest(
                String destination,
                T request,
                Map<String, String> headers,
                Class<U> replyType); Single<RpcReply<U>> sendDeleteRequest(
                String destination,
                T request,
                Class<U> replyType,
                Duration timeout); Single<RpcReply<U>> sendDeleteRequest(
                String destination,
                T request,
                Map<String, String> headers,
                Class<U> replyType,
                Duration timeout); Single<RpcReply<U>> sendDeleteRequest(
                String destination,
                T request,
                Function<String, U> replyTranscriber); Single<RpcReply<U>> sendDeleteRequest(
                String destination,
                T request,
                Map<String, String> headers,
                Function<String, U> replyTranscriber); Single<RpcReply<U>> sendDeleteRequest(
                String destination,
                T request,
                Function<String, U> replyTranscriber,
                Duration timeout); Single<RpcReply<U>> sendDeleteRequest(
                String destination,
                T request,
                Map<String, String> headers,
                Function<String, U> replyTranscriber,
                Duration timeout); Single<RpcReply<InputStream>> sendDeleteRequestAndRetrieveResponseAsStream(
                String destination,
                T request); Single<RpcReply<InputStream>> sendDeleteRequestAndRetrieveResponseAsStream(
                String destination,
                T request,
                Map<String, String> headers); Single<RpcReply<InputStream>> sendDeleteRequestAndRetrieveResponseAsStream(
                String destination,
                T request,
                Duration timeout); Single<RpcReply<InputStream>> sendDeleteRequestAndRetrieveResponseAsStream(
                String destination,
                T request,
                Map<String, String> headers,
                Duration timeout); Single<RpcReply<U>> sendRequest(
                String destination,
                T request,
                HttpRequestMethod requestMethod,
                Class<U>replyType); Single<RpcReply<U>> sendRequest(
                String destination,
                T request,
                HttpRequestMethod requestMethod,
                Map<String, String> headers,
                Class<U>replyType); Single<RpcReply<U>> sendRequest(
                String destination,
                T request,
                HttpRequestMethod requestMethod,
                Class<U> replyType,
                Duration timeout); Single<RpcReply<U>> sendRequest(
                String destination,
                T request,
                HttpRequestMethod requestMethod,
                Map<String, String> headers,
                Class<U> replyType,
                Duration timeout); Single<RpcReply<U>> sendRequest(
                String destination,
                T request,
                HttpRequestMethod requestMethod,
                Function<String, U> replyTranscriber); Single<RpcReply<U>> sendRequest(
                String destination,
                T request,
                HttpRequestMethod requestMethod,
                Map<String, String> headers,
                Function<String, U> replyTranscriber); Single<RpcReply<U>> sendRequest(
                String destination,
                T request,
                HttpRequestMethod requestMethod,
                Function<String, U> replyTranscriber,
                Duration timeout); Single<RpcReply<U>> sendRequest(
                String destination,
                T request,
                HttpRequestMethod requestMethod,
                Map<String, String> headers,
                Function<String, U> replyTranscriber,
                Duration timeout); Single<RpcReply<InputStream>> sendRequestAndRetrieveResponseAsStream(
                String destination,
                T request,
                HttpRequestMethod requestMethod); Single<RpcReply<InputStream>> sendRequestAndRetrieveResponseAsStream(
                String destination,
                T request,
                HttpRequestMethod requestMethod,
                Map<String, String> headers); Single<RpcReply<InputStream>> sendRequestAndRetrieveResponseAsStream(
                String destination,
                T request,
                HttpRequestMethod requestMethod,
                Duration timeout); Single<RpcReply<InputStream>> sendRequestAndRetrieveResponseAsStream(
                String destination,
                T request,
                HttpRequestMethod requestMethod,
                Map<String, String> headers,
                Duration timeout); Single<RpcReply<U>> sendRequest(
                String destination,
                T request,
                RequestInfo sendingInfo,
                Class<U> replyType,
                Duration timeout); Single<RpcReply<U>> sendRequest(
                String destination,
                T request,
                RequestInfo httpInfo,
                Function<String, U> replyTranscriber,
                Duration timeout); Single<RpcReply<InputStream>> sendRequestAndRetrieveResponseAsStream(
                String destination,
                T request,
                RequestInfo httpInfo,
                Duration timeout); Flowable<RpcInvocation<T>> requests(String destination, Class<T> requestType); void start(); void shutdown(); @Override void httpServerInstanceCreated(HttpServer httpServer); @Override void httpClientInstanceCreated(AsyncHttpClient httpClient); static Builder builder();  }
@Test void sameAdapterCanBeUsedForDifferentMessageTypes() { String path = "/multiTypeTest"; Pair<com.sun.net.httpserver.HttpServer, Integer> serverPortPair = httpServer(path, "1"); String url = "http: TestObserver<RpcReply<String>> stringObserver = sut.sendGetRequest(url, String.class).test(); TestObserver<RpcReply<Integer>> integerObserver = sut.sendGetRequest(url, Integer.class).test(); TestObserver<RpcReply<Double>> doubleObserver = sut.sendGetRequest(url, Double.class).test(); TestObserver<RpcReply<BigDecimal>> bigDecimalObserver = sut.sendGetRequest(url, BigDecimal.class).test(); await().atMost(10, SECONDS).until(stringObserver::isTerminated, is(true)); stringObserver.assertValue(reply -> "1".equals(reply.result)); await().atMost(10, SECONDS).until(integerObserver::isTerminated, is(true)); integerObserver.assertValue(reply -> 1 == reply.result); await().atMost(10, SECONDS).until(doubleObserver::isTerminated, is(true)); doubleObserver.assertValue(reply -> 1.0 == reply.result); await().atMost(10, SECONDS).until(bigDecimalObserver::isTerminated, is(true)); bigDecimalObserver.assertValue(reply -> BigDecimal.ONE.equals(reply.result)); }
@Test void oneOffTranscriberCanBeUsed() { String path = "/jacksonTest"; Pair<com.sun.net.httpserver.HttpServer, Integer> serverPortPair = httpServer(path, "xxx"); String url = "http: Function<String, MyType1> replyTranscriber = string -> { throw new RuntimeException("Oli"); }; TestObserver<RpcReply<MyType1>> type1Observer = sut.sendGetRequest(url, replyTranscriber).test(); type1Observer.assertError(throwable -> throwable instanceof RuntimeException && throwable.getMessage().equals("Oli")); }
@Test void parametersGetProperlyTransmitted() { Map<String, String> headers = new HashMap<>(); headers.put("h1", "v1"); headers.put("h2", "v2"); Headers[] headersContainer = new Headers[1]; String path = "/paramTest"; Pair<com.sun.net.httpserver.HttpServer, Integer> serverPortPair = httpServer( path, "xxx", httpExchange -> { headersContainer[0] = httpExchange.getRequestHeaders(); }); String url = "http: sut.sendGetRequest(url, headers, String.class).subscribe(); await().atMost(10, SECONDS).until(() -> headersContainer[0], not(nullValue())); MatcherAssert.assertThat(headersContainer[0].getFirst("h1"), is("v1")); MatcherAssert.assertThat(headersContainer[0].getFirst("h2"), is("v2")); }