Create new async client based on Reactor Netty

build.gradle

@@ -37,8 +37,9 @@ ext {
   groovyVersion = '2.4.13'
   jacksonVersion = '2.9.4'
   spockVersion = "1.1-groovy-2.4"
-  reactorVersion = "Bismuth-SR3"
+  reactorVersion = "Bismuth-SR8"
   springVersion = '5.0.3.RELEASE'
+  reactorNettyVersion = '0.7.6.RELEASE'
 
   linkHomepage = 'https://github.com/rabbitmq/hop'
   linkCi       = 'https://build.spring.io/browse/RMQ'
@@ -63,7 +64,7 @@ dependencies {
   compile "org.springframework:spring-webflux:$springVersion"
   compile "org.apache.httpcomponents:httpclient:$commonsHttpClientVersion"
   compile "com.fasterxml.jackson.core:jackson-databind:$jacksonVersion"
-  compile "io.projectreactor.ipc:reactor-netty"
+  compile "io.projectreactor.ipc:reactor-netty:$reactorNettyVersion"
 
   testCompile "org.codehaus.groovy:groovy-all:$groovyVersion"
   testCompile "org.spockframework:spock-core:$spockVersion"

src/main/java/com/rabbitmq/http/client/ReactorNettyClient.java

@@ -0,0 +1,202 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.rabbitmq.http.client;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.MapperFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.rabbitmq.http.client.domain.NodeInfo;
+import com.rabbitmq.http.client.domain.OverviewResponse;
+import com.rabbitmq.http.client.domain.PolicyInfo;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpHeaderValues;
+import io.netty.handler.codec.json.JsonObjectDecoder;
+import org.reactivestreams.Publisher;
+import org.springframework.util.StringUtils;
+import reactor.core.Exceptions;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.ipc.netty.http.client.HttpClient;
+import reactor.ipc.netty.http.client.HttpClientRequest;
+import reactor.ipc.netty.http.client.HttpClientResponse;
+
+import java.io.UnsupportedEncodingException;
+import java.lang.reflect.Array;
+import java.net.URI;
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.function.Function;
+
+/**
+ *
+ */
+public class ReactorNettyClient {
+
+    private static final int MAX_PAYLOAD_SIZE = 100 * 1024 * 1024;
+
+    private static final String ENCODING_CHARSET = "UTF-8";
+
+    private final String rootUrl;
+
+    private final ObjectMapper objectMapper = new ObjectMapper();
+
+    private final HttpClient client;
+
+    private final Mono<String> token;
+
+    public ReactorNettyClient(String url) {
+        this(urlWithoutCredentials(url),
+            StringUtils.split(URI.create(url).getUserInfo(), ":")[0],
+            StringUtils.split(URI.create(url).getUserInfo(), ":")[1]);
+    }
+
+    public ReactorNettyClient(String url, String username, String password) {
+        rootUrl = url;
+        // FIXME make Jackson ObjectMapper configurable
+        objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
+        objectMapper.enable(DeserializationFeature.ACCEPT_EMPTY_ARRAY_AS_NULL_OBJECT);
+        objectMapper.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
+        objectMapper.disable(MapperFeature.DEFAULT_VIEW_INCLUSION);
+
+        // FIXME make URL configurable
+        URI uri = URI.create(url);
+        client = HttpClient.create(options -> options.host(uri.getHost()).port(uri.getPort()));
+
+        // FIXME make Authentication header value configurable (default being Basic)
+        this.token = createBasicAuthenticationToken(username, password);
+
+        // FIXME make SSLContext configurable when using TLS
+    }
+
+    private static String urlWithoutCredentials(String url) {
+        URI url1 = URI.create(url);
+        return StringUtils.replace(url, url1.getUserInfo() + "@", "");
+    }
+
+    protected Mono<String> createBasicAuthenticationToken(String username, String password) {
+        return Mono.fromSupplier(() -> {
+            String credentials = username + ":" + password;
+            byte[] credentialsAsBytes = credentials.getBytes(StandardCharsets.ISO_8859_1);
+            byte[] encodedBytes = Base64.getEncoder().encode(credentialsAsBytes);
+            String encodedCredentials = new String(encodedBytes, StandardCharsets.ISO_8859_1);
+            return "Basic " + encodedCredentials;
+        }).cache();
+    }
+
+    public Mono<OverviewResponse> getOverview() {
+        return doGetMono(OverviewResponse.class, "overview");
+    }
+
+    public Flux<NodeInfo> getNodes() {
+        return doGetFlux(NodeInfo.class, "nodes");
+    }
+
+    public Mono<HttpClientResponse> declarePolicy(String vhost, String name, PolicyInfo info) {
+        return doPost(info, "policies", vhost, name);
+    }
+
+    private HttpClientRequest disableChunkTransfer(HttpClientRequest request) {
+        return request.chunkedTransfer(false);
+    }
+
+    public Flux<PolicyInfo> getPolicies() {
+        return doGetFlux(PolicyInfo.class, "policies");
+    }
+
+    public Mono<HttpClientResponse> deletePolicy(String vhost, String name) {
+        return doDelete("policies", vhost, name);
+    }
+
+    private <T> Mono<T> doGetMono(Class<T> type, String... pathSegments) {
+        return client.get(uri(pathSegments), request -> Mono.just(request)
+            .transform(this::addAuthorization)
+            .flatMap(pRequest -> pRequest.send())).transform(decode(type));
+    }
+
+    private <T> Flux<T> doGetFlux(Class<T> type, String... pathSegments) {
+        return (Flux<T>) doGetMono(Array.newInstance(type, 0).getClass(), pathSegments).flatMapMany(items -> Flux.fromArray((Object[]) items));
+    }
+
+    private Mono<HttpClientResponse> doPost(Object body, String... pathSegments) {
+        return client.put(uri(pathSegments), request -> Mono.just(request)
+            .transform(this::addAuthorization)
+            .map(this::disableChunkTransfer)
+            .transform(encode(body)));
+    }
+
+    private Mono<HttpClientResponse> doDelete(String... pathSegments) {
+        return client.delete(uri(pathSegments), request -> Mono.just(request)
+            .transform(this::addAuthorization)
+            .flatMap(HttpClientRequest::send)
+        );
+    }
+
+    private Mono<HttpClientRequest> addAuthorization(Mono<HttpClientRequest> request) {
+        return Mono
+            .zip(request, token)
+            .map(tuple -> tuple.getT1().addHeader(HttpHeaderNames.AUTHORIZATION, tuple.getT2()));
+    }
+
+    private String uri(String... pathSegments) {
+        StringBuilder builder = new StringBuilder();
+        if (pathSegments != null && pathSegments.length > 0) {
+            for (String pathSegment : pathSegments) {
+                try {
+                    builder.append("/");
+                    builder.append(URLEncoder.encode(pathSegment, ENCODING_CHARSET));
+                } catch (UnsupportedEncodingException e) {
+                    // FIXME exception handling
+                    throw new RuntimeException(e);
+                }
+            }
+        }
+        return rootUrl + builder.toString();
+    }
+
+    private <T> Function<Mono<HttpClientResponse>, Flux<T>> decode(Class<T> type) {
+        return inbound ->
+            inbound.flatMapMany(response -> response.addHandler(new JsonObjectDecoder(MAX_PAYLOAD_SIZE)).receive().asByteArray()
+                .map(payload -> {
+                    try {
+                        return objectMapper.readValue(payload, type);
+                    } catch (Throwable t) {
+                        // FIXME exception handling
+                        throw new RuntimeException(t);
+                    }
+                })
+            );
+    }
+
+    private Function<Mono<HttpClientRequest>, Publisher<Void>> encode(Object requestPayload) {
+        return outbound -> outbound
+            .flatMapMany(request -> {
+                try {
+                    byte[] bytes = objectMapper.writeValueAsBytes(requestPayload);
+
+                    return request
+                        .header(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON)
+                        .header(HttpHeaderNames.CONTENT_LENGTH, String.valueOf(bytes.length))
+                        .sendByteArray(Mono.just(bytes));
+                } catch (JsonProcessingException e) {
+                    throw Exceptions.propagate(e);
+                }
+            });
+    }
+}

src/test/groovy/com/rabbitmq/http/client/ReactorNettyClientSpec.groovy

@@ -0,0 +1,165 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.rabbitmq.http.client
+
+import com.rabbitmq.client.Connection
+import com.rabbitmq.client.ConnectionFactory
+import com.rabbitmq.http.client.domain.NodeInfo
+import com.rabbitmq.http.client.domain.PolicyInfo
+import spock.lang.Specification
+
+class ReactorNettyClientSpec extends Specification {
+
+    protected ReactorNettyClient client
+
+    private final ConnectionFactory cf = initializeConnectionFactory()
+
+    protected static ConnectionFactory initializeConnectionFactory() {
+        final cf = new ConnectionFactory()
+        cf.setAutomaticRecoveryEnabled(false)
+        cf
+    }
+
+    def setup() {
+        client = newLocalhostNodeClient()
+    }
+
+    protected static ReactorNettyClient newLocalhostNodeClient() {
+        new ReactorNettyClient("http://guest:guest@localhost:15672/api")
+    }
+
+    def "GET /api/overview"() {
+        when: "client requests GET /api/overview"
+        final conn = openConnection()
+        final ch = conn.createChannel()
+        1000.times { ch.basicPublish("", "", null, null) }
+
+        def res = client.getOverview().block()
+        def xts = res.getExchangeTypes().collect { it.getName() }
+
+        then: "the response is converted successfully"
+        res.getNode().startsWith("rabbit@")
+        res.getErlangVersion() != null
+
+        final msgStats = res.getMessageStats()
+        msgStats.basicPublish >= 0
+        msgStats.publisherConfirm >= 0
+        msgStats.basicDeliver >= 0
+        msgStats.basicReturn >= 0
+
+        final qTotals = res.getQueueTotals()
+        qTotals.messages >= 0
+        qTotals.messagesReady >= 0
+        qTotals.messagesUnacknowledged >= 0
+
+        final oTotals = res.getObjectTotals()
+        oTotals.connections >= 0
+        oTotals.channels >= 0
+        oTotals.exchanges >= 0
+        oTotals.queues >= 0
+        oTotals.consumers >= 0
+
+        res.listeners.size() >= 1
+        res.contexts.size() >= 1
+
+        xts.contains("topic")
+        xts.contains("fanout")
+        xts.contains("direct")
+        xts.contains("headers")
+
+        cleanup:
+        if (conn.isOpen()) {
+            conn.close()
+        }
+    }
+
+    def "GET /api/nodes"() {
+        when: "client retrieves a list of cluster nodes"
+        final res = client.getNodes()
+        final node = res.blockFirst()
+
+        then: "the list is returned"
+        res.count().block() >= 1
+        verifyNode(node)
+    }
+
+    def "GET /api/policies"() {
+        given: "at least one policy was declared"
+        final v = "/"
+        final s = "hop.test"
+        final d = new HashMap<String, Object>()
+        final p = ".*"
+        d.put("ha-mode", "all")
+        client.declarePolicy(v, s, new PolicyInfo(p, 0, null, d)).block()
+
+        when: "client lists policies"
+        final xs = awaitEventPropagation({ client.getPolicies() })
+
+        then: "a list of policies is returned"
+        final x = xs.blockFirst()
+        verifyPolicyInfo(x)
+
+        cleanup:
+        client.deletePolicy(v, s).block()
+    }
+
+    protected Connection openConnection() {
+        this.cf.newConnection()
+    }
+
+    protected static void verifyNode(NodeInfo node) {
+        assert node != null
+        assert node.name != null
+        assert node.socketsUsed <= node.socketsTotal
+        assert node.erlangProcessesUsed <= node.erlangProcessesTotal
+        assert node.erlangRunQueueLength >= 0
+        assert node.memoryUsed <= node.memoryLimit
+    }
+
+    protected static void verifyPolicyInfo(PolicyInfo x) {
+        assert x.name != null
+        assert x.vhost != null
+        assert x.pattern != null
+        assert x.definition != null
+        assert x.applyTo != null
+    }
+
+    /**
+     * Statistics tables in the server are updated asynchronously,
+     * in particular starting with rabbitmq/rabbitmq-management#236,
+     * so in some cases we need to wait before GET'ing e.g. a newly opened connection.
+     */
+    protected static Object awaitEventPropagation(Closure callback) {
+        if (callback) {
+            int n = 0
+            def result = callback()
+            def hasElements = false
+            while (!hasElements && n < 10000) {
+                Thread.sleep(100)
+                n += 100
+                result = callback()
+                hasElements = result?.hasElements().block()
+            }
+            assert n < 10000
+            result
+        } else {
+            Thread.sleep(1000)
+            null
+        }
+    }
+
+}