update to latest rsocket-rpc-java and reactor versions (#18)

Author: OlegDokuka

gradle.properties

@@ -1,2 +1,2 @@
 group=io.netifi.proteus
-version=0.9.4
+version=0.9.5

gradle/java.gradle

@@ -35,7 +35,7 @@ task ciVersion {
 build.dependsOn ciVersion
 
 ext {
-    rsocketRpcVersion = '0.2.2'
+    rsocketRpcVersion = '0.2.3'
     protobufVersion = '3.6.1'
 }
 
@@ -88,14 +88,14 @@ dependencies {
     testCompile 'junit:junit:4.12'
 
     testCompile 'javax.inject:javax.inject:1'
-    testCompile 'io.projectreactor:reactor-test:3.1.9.RELEASE'
+    testCompile 'io.projectreactor:reactor-test:3.2.0.RELEASE'
     testCompile "com.google.protobuf:protobuf-java:$protobufVersion"
     testCompile 'org.hdrhistogram:HdrHistogram:2.1.10'
     testCompile 'org.apache.logging.log4j:log4j-api:2.9.0'
     testCompile 'org.apache.logging.log4j:log4j-core:2.9.0'
     testCompile 'org.apache.logging.log4j:log4j-slf4j-impl:2.9.0'
-    testCompile 'io.rsocket:rsocket-transport-netty:0.11.7'
-    testCompile 'io.rsocket:rsocket-transport-local:0.11.7'
+    testCompile 'io.rsocket:rsocket-transport-netty:0.11.8'
+    testCompile 'io.rsocket:rsocket-transport-local:0.11.8'
     testCompile 'org.mockito:mockito-all:1.10.19'
 }
 

proteus-client/src/main/java/io/netifi/proteus/Proteus.java

@@ -26,7 +26,7 @@
 import reactor.core.Exceptions;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoProcessor;
-import reactor.ipc.netty.tcp.TcpClient;
+import reactor.netty.tcp.TcpClient;
 
 /** This is where the magic happens */
 public class Proteus implements Closeable {
@@ -256,7 +256,7 @@ public Proteus build() {
         if (sslDisabled) {
           clientTransportFactory =
               address -> {
-                TcpClient client = TcpClient.create(opts -> opts.connectAddress(() -> address));
+                TcpClient client = TcpClient.create().addressSupplier(() -> address);
                 return TcpClientTransport.create(client);
               };
         } else {
@@ -277,8 +277,7 @@ public Proteus build() {
             clientTransportFactory =
                 address -> {
                   TcpClient client =
-                      TcpClient.create(
-                          opts -> opts.connectAddress(() -> address).sslContext(sslContext));
+                      TcpClient.create().addressSupplier(() -> address).secure(sslContext);
                   return TcpClientTransport.create(client);
                 };
           } catch (Exception sslException) {

proteus-metrics-prometheus/src/main/java/io/netifi/proteus/prometheus/ProteusPrometheusBridge.java

@@ -26,9 +26,9 @@
 import reactor.core.publisher.DirectProcessor;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.ipc.netty.http.server.HttpServer;
-import reactor.ipc.netty.http.server.HttpServerRequest;
-import reactor.ipc.netty.http.server.HttpServerResponse;
+import reactor.netty.http.server.HttpServer;
+import reactor.netty.http.server.HttpServerRequest;
+import reactor.netty.http.server.HttpServerResponse;
 
 @Named("ProteusPrometheusBridge")
 public class ProteusPrometheusBridge implements MetricsSnapshotHandler {
@@ -99,8 +99,11 @@ public static void main(String... args) {
   }
 
   private void init() {
-    HttpServer.create(bindAddress, bindPort)
-        .newRouter(routes -> routes.post(metricsUrl, this::handle).get(metricsUrl, this::handle))
+    HttpServer.create()
+        .host(bindAddress)
+        .port(bindPort)
+        .route(routes -> routes.post(metricsUrl, this::handle).get(metricsUrl, this::handle))
+        .bind()
         .subscribe();
   }
 

proteus-tracing-openzipkin/build.gradle

@@ -10,7 +10,7 @@ dependencies {
     protobuf project (':proteus-tracing-idl')
     compile project (':proteus-client')
 
-    compile 'io.projectreactor.addons:reactor-adapter:3.1.7.RELEASE'
+    compile 'io.projectreactor.addons:reactor-adapter:3.2.0.RELEASE'
     compile "com.google.protobuf:protobuf-java-util:$protobufVersion"
 
     compile 'io.opentracing:opentracing-api:0.31.0'

proteus-tracing-openzipkin/src/main/java/io/netifi/proteus/tracing/ProteusZipkinHttpBridge.java

@@ -13,8 +13,9 @@
 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.resources.PoolResources;
+import reactor.netty.ByteBufFlux;
+import reactor.netty.http.client.HttpClient;
+import reactor.netty.resources.ConnectionProvider;
 import zipkin2.proto3.Span;
 
 public class ProteusZipkinHttpBridge implements ProteusTracingService {
@@ -82,17 +83,15 @@ public static void main(String... args) {
   private synchronized HttpClient getClient() {
     if (httpClient == null) {
       this.httpClient =
-          HttpClient.builder()
-              .options(
-                  builder ->
-                      builder
-                          .compression(true)
-                          .poolResources(PoolResources.fixed("proteusZipkinBridge"))
-                          .option(ChannelOption.SO_KEEPALIVE, true)
-                          .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 30_000)
+          HttpClient.create(ConnectionProvider.fixed("proteusZipkinBridge"))
+              .compress(true)
+              .port(port)
+              .tcpConfiguration(
+                  tcpClient ->
+                      tcpClient
                           .host(host)
-                          .port(port))
-              .build();
+                          .option(ChannelOption.SO_KEEPALIVE, true)
+                          .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 30_000));
     }
     return httpClient;
   }
@@ -127,12 +126,11 @@ public Mono<Ack> streamSpans(Publisher<Span> messages, ByteBuf metadata) {
         .concatMap(
             spans ->
                 getClient()
-                    .post(
-                        zipkinSpansUrl,
-                        request -> {
-                          request.addHeader("Content-Type", "application/json");
-                          return request.sendString(Mono.just(spans));
-                        })
+                    .headers(hh -> hh.add("Content-Type", "application/json"))
+                    .post()
+                    .uri(zipkinSpansUrl)
+                    .send(ByteBufFlux.fromString(Mono.just(spans)))
+                    .response()
                     .timeout(Duration.ofSeconds(30))
                     .doOnError(throwable -> resetHttpClient()),
             8)

proteus-tracing-openzipkin/src/main/java/io/netifi/proteus/tracing/TracesStreamer.java

@@ -17,7 +17,7 @@
 import reactor.core.Exceptions;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.ipc.netty.http.client.HttpClient;
+import reactor.netty.http.client.HttpClient;
 import zipkin2.proto3.Span;
 
 public class TracesStreamer {
@@ -66,13 +66,12 @@ private static Function<Integer, Publisher<InputStream>> zipkinServerStream(
     return lookbackSeconds ->
         client.flatMapMany(
             c ->
-                c.get(
-                        zipkinQuery(zipkinUrl, lookbackSeconds),
-                        req -> {
-                          req.context().addHandler(new JsonObjectDecoder(true));
-                          return Mono.empty();
-                        })
-                    .flatMapMany(resp -> resp.receive().asInputStream()));
+                c.doOnRequest(
+                        (__, connection) -> connection.addHandler(new JsonObjectDecoder(true)))
+                    .get()
+                    .uri(zipkinQuery(zipkinUrl, lookbackSeconds))
+                    .responseContent()
+                    .asInputStream());
   }
 
   private static String zipkinQuery(String zipkinUrl, int lookbackSeconds) {

proteus-tracing-openzipkin/src/test/java/io/netifi/proteus/tracing/ZipkinTracesStreamerTest.java

@@ -14,8 +14,8 @@
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.ipc.netty.http.client.HttpClient;
-import reactor.ipc.netty.resources.PoolResources;
+import reactor.netty.http.client.HttpClient;
+import reactor.netty.resources.ConnectionProvider;
 
 public class ZipkinTracesStreamerTest {
 
@@ -70,16 +70,14 @@ public void streamerIntegrationTest() {
 
   private Mono<HttpClient> client() {
     return Mono.just(
-        HttpClient.builder()
-            .options(
-                builder ->
-                    builder
-                        .compression(true)
-                        .poolResources(PoolResources.fixed("proteusZipkinBridge"))
+        HttpClient.create(ConnectionProvider.fixed("proteusZipkinBridge"))
+            .compress(true)
+            .port(9411)
+            .tcpConfiguration(
+                tcpClient ->
+                    tcpClient
                         .option(ChannelOption.SO_KEEPALIVE, true)
                         .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 30_000)
-                        .host("127.0.0.1")
-                        .port(9411))
-            .build());
+                        .host("127.0.0.1")));
   }
 }