Merge branch 'develop' of github.com:netifi-proteus/proteus-java into migrate-to-rsocket-rpc

Author: kbahr

README.md

@@ -1,6 +1,6 @@
 # Netifi Proteus Java
 
-<a href='https://travis-ci.org/netifi-proteus/proteus-java'><img src='https://travis-ci.org/netifi-proteus/proteus-java.svg?branch=master'></a>
+[![Join the chat at https://gitter.im/netifi/general](https://badges.gitter.im/netifi/general.svg)](https://gitter.im/netifi/general) <a href='https://travis-ci.org/netifi-proteus/proteus-java'><img src='https://travis-ci.org/netifi-proteus/proteus-java.svg?branch=master'></a>
 
 
 ## Build from Source

client/build.gradle

@@ -15,6 +15,7 @@ dependencies {
 
     compile 'com.typesafe:config:1.3.2'
     compile 'com.google.guava:guava:22.0'
+    compile 'io.netty:netty-tcnative:2.0.12.Final:linux-x86_64'
 
     testCompile 'io.micrometer:micrometer-registry-atlas:1.0.3'
     testProtobuf 'io.rsocket.rpc:core-idl:0.1.2'

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

@@ -1,19 +1,25 @@
 package io.netifi.proteus;
 
-import io.micrometer.core.instrument.MeterRegistry;
 import io.netifi.proteus.rsocket.ProteusSocket;
 import io.rsocket.rpc.rsocket.RequestHandlingRSocket;
 import io.rsocket.rpc.RSocketRpcService;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
+import io.netty.handler.ssl.OpenSsl;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslProvider;
+import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
 import io.opentracing.Tracer;
 import io.rsocket.Closeable;
 import io.rsocket.transport.ClientTransport;
 import io.rsocket.transport.netty.client.TcpClientTransport;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import reactor.core.Exceptions;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoProcessor;
+import reactor.ipc.netty.tcp.TcpClient;
 
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
@@ -135,10 +141,7 @@ public static class Builder {
     private int missedAcks = DefaultBuilderConfig.getMissedAcks();
     private DestinationNameFactory destinationNameFactory;
 
-    private MeterRegistry registry = null;
-    private int batchSize = DefaultBuilderConfig.getBatchSize();
-    private Function<SocketAddress, ClientTransport> clientTransportFactory =
-        address -> TcpClientTransport.create((InetSocketAddress) address);
+    private Function<SocketAddress, ClientTransport> clientTransportFactory = null;
     private int poolSize = Runtime.getRuntime().availableProcessors();
     private Supplier<Tracer> tracerSupplier = () -> null;
 
@@ -153,13 +156,8 @@ public Builder poolSize(int poolSize) {
       return this;
     }
 
-    public Builder metricBatchSize(int batchSize) {
-      this.batchSize = batchSize;
-      return this;
-    }
-
     public Builder keepalive(boolean useKeepAlive) {
-      this.keepalive = keepalive;
+      this.keepalive = useKeepAlive;
       return this;
     }
 
@@ -197,9 +195,7 @@ public Builder seedAddresses(Collection<SocketAddress> addresses) {
       if (addresses instanceof List) {
         this.seedAddresses = (List<SocketAddress>) addresses;
       } else {
-        List<SocketAddress> list = new ArrayList<>();
-        list.addAll(addresses);
-        this.seedAddresses = list;
+        this.seedAddresses = new ArrayList<>(addresses);
       }
 
       return this;
@@ -246,6 +242,35 @@ public Proteus build() {
       Objects.requireNonNull(accessToken, "account token is required");
       Objects.requireNonNull(group, "group is required");
 
+      if (clientTransportFactory == null) {
+        logger.info("Client transport factory not provided; using TCP transport.");
+        try {
+          final SslProvider sslProvider;
+          if (OpenSsl.isAvailable()) {
+            logger.info("Native SSL provider is available; will use native provider.");
+            sslProvider = SslProvider.OPENSSL_REFCNT;
+          } else {
+            logger.info("Native SSL provider not available; will use JDK SSL provider.");
+            sslProvider = SslProvider.JDK;
+          }
+          final SslContext sslContext =
+              SslContextBuilder.forClient()
+                  .trustManager(InsecureTrustManagerFactory.INSTANCE)
+                  .sslProvider(sslProvider)
+                  .build();
+          clientTransportFactory = address -> {
+            TcpClient client =
+                TcpClient.create(
+                    opts ->
+                        opts.connectAddress(() -> address)
+                            .sslContext(sslContext));
+            return TcpClientTransport.create(client);
+          };
+        } catch (Exception sslException) {
+          throw Exceptions.bubble(sslException);
+        }
+      }
+
       this.accessTokenBytes = Base64.getDecoder().decode(accessToken);
 
       if (destinationNameFactory == null) {
@@ -260,7 +285,7 @@ public Proteus build() {
       if (seedAddresses == null) {
         Objects.requireNonNull(host, "host is required");
         Objects.requireNonNull(port, "port is required");
-        socketAddresses = Arrays.asList(InetSocketAddress.createUnresolved(host, port));
+        socketAddresses = Collections.singletonList(InetSocketAddress.createUnresolved(host, port));
       } else {
         socketAddresses = seedAddresses;
       }