Refactor Factory to Connector + Implement availability (#15)

* Refactor Factory to Connector + Implement availability

Problem
We need to comply with the recent refactoring "Factory to Connector" introduced
in reactivesocket-java. Also, implementation of DuplexConnection have to return
an availability.

Solution
The Factory to Connector refactoring is pretty straight forward.

All implementations of DuplexConnection return an availability which directly
map the state of the underlying resource (0.0 if the resource is closed, 1.0
otherwise). This will greatly help the load-balancer to select a valid
connection.

I removed the blocking method from reactivesocket-java, so I moved some utility
blocking code inside the TestUtil class.

Clean-up of the Netty implementation, remove unused args, explicitey specify
tcp socket configuration.

Shortened the toString name to ease log-reading.

Bug
All DuplexConnection implementations now cancel the subscription when an exception
occurs. This fix the problem, where the Keep-Alive Observable kept trying to
send a keep-alive on a closed connection.

* Refactor the TestUtil helper to return CompletableFuture

* rs version

* Restore rs dependency to latest.release

* Remove mavenLocal in gradle build file

* Propagate cancellation in all cases

* Use helper method from ReactiveSocket Unsafe

Author: stevegury

reactivesocket-aeron/src/main/java/io/reactivesocket/aeron/client/AeronClientDuplexConnection.java

@@ -118,6 +118,11 @@ public void onComplete() {
             });
     }
 
+    @Override
+    public double availability() {
+        return publication.isClosed() ? 0.0 : 1.0;
+    }
+
     @Override
     public void close() throws IOException {
         onClose.accept(publication);
@@ -136,6 +141,5 @@ public String toString() {
             "channel=" + publication.channel() + "," +
             "streamId=" + publication.streamId() + "," +
             "sessionId=" + publication.sessionId() + "]";
-
     }
 }

reactivesocket-aeron/src/main/java/io/reactivesocket/aeron/client/AeronReactiveSocketFactory.java renamed to reactivesocket-aeron/src/main/java/io/reactivesocket/aeron/client/AeronReactiveSocketConnector.java

@@ -15,10 +15,7 @@
  */
 package io.reactivesocket.aeron.client;
 
-import io.reactivesocket.ConnectionSetupPayload;
-import io.reactivesocket.DefaultReactiveSocket;
-import io.reactivesocket.ReactiveSocket;
-import io.reactivesocket.ReactiveSocketFactory;
+import io.reactivesocket.*;
 import io.reactivesocket.rx.Completable;
 import org.agrona.LangUtil;
 import org.reactivestreams.Publisher;
@@ -40,17 +37,17 @@
 /**
  * An implementation of {@link ReactiveSocketFactory} that creates Aeron ReactiveSockets.
  */
-public class AeronReactiveSocketFactory implements ReactiveSocketFactory<SocketAddress, ReactiveSocket> {
-    private static final Logger logger = LoggerFactory.getLogger(AeronReactiveSocketFactory.class);
+public class AeronReactiveSocketConnector implements ReactiveSocketConnector<SocketAddress> {
+    private static final Logger logger = LoggerFactory.getLogger(AeronReactiveSocketConnector.class);
 
     private final ConnectionSetupPayload connectionSetupPayload;
     private final Consumer<Throwable> errorStream;
 
-    public AeronReactiveSocketFactory(ConnectionSetupPayload connectionSetupPayload, Consumer<Throwable> errorStream) {
+    public AeronReactiveSocketConnector(ConnectionSetupPayload connectionSetupPayload, Consumer<Throwable> errorStream) {
         this(getIPv4InetAddress().getHostAddress(), 39790, connectionSetupPayload, errorStream);
     }
 
-    public AeronReactiveSocketFactory(String host, int port, ConnectionSetupPayload connectionSetupPayload, Consumer<Throwable> errorStream) {
+    public AeronReactiveSocketConnector(String host, int port, ConnectionSetupPayload connectionSetupPayload, Consumer<Throwable> errorStream) {
         this.connectionSetupPayload = connectionSetupPayload;
         this.errorStream = errorStream;
 
@@ -65,7 +62,7 @@ public AeronReactiveSocketFactory(String host, int port, ConnectionSetupPayload
     }
 
     @Override
-    public Publisher<ReactiveSocket> call(SocketAddress address) {
+    public Publisher<ReactiveSocket> connect(SocketAddress address) {
         Publisher<AeronClientDuplexConnection> connection
             = AeronClientDuplexConnectionFactory.getInstance().createAeronClientDuplexConnection(address);
 

reactivesocket-aeron/src/main/java/io/reactivesocket/aeron/server/AeronServerDuplexConnection.java

@@ -78,6 +78,11 @@ public void addOutput(Publisher<Frame> o, Completable callback) {
         o.subscribe(new ServerSubscription(publication, callback));
     }
 
+    @Override
+    public double availability() {
+        return isClosed ? 0.0 : 1.0;
+    }
+
     // TODO - this is bad - I need to queue this up somewhere and process this on the polling thread so it doesn't just block everything
     void ackEstablishConnection(int ackSessionId) {
         debug("Acking establish connection for session id => {}", ackSessionId);

reactivesocket-jsr-356/src/main/java/io/reactivesocket/javax/websocket/WebSocketDuplexConnection.java

@@ -79,6 +79,11 @@ public void onSubscribe(Subscription s) {
         });
     }
 
+    @Override
+    public double availability() {
+        return session.isOpen() ? 1.0 : 0.0;
+    }
+
     @Override
     public void close() throws IOException {
         session.close();

reactivesocket-jsr-356/src/main/java/io/reactivesocket/javax/websocket/client/WebSocketReactiveSocketFactory.java renamed to reactivesocket-jsr-356/src/main/java/io/reactivesocket/javax/websocket/client/WebSocketReactiveSocketConnector.java

@@ -15,10 +15,7 @@
  */
 package io.reactivesocket.javax.websocket.client;
 
-import io.reactivesocket.ConnectionSetupPayload;
-import io.reactivesocket.DefaultReactiveSocket;
-import io.reactivesocket.ReactiveSocket;
-import io.reactivesocket.ReactiveSocketFactory;
+import io.reactivesocket.*;
 import io.reactivesocket.javax.websocket.WebSocketDuplexConnection;
 import io.reactivesocket.rx.Completable;
 import org.glassfish.tyrus.client.ClientManager;
@@ -36,23 +33,23 @@
 /**
  * An implementation of {@link ReactiveSocketFactory} that creates JSR-356 WebSocket ReactiveSockets.
  */
-public class WebSocketReactiveSocketFactory implements ReactiveSocketFactory<SocketAddress, ReactiveSocket> {
-    private static final Logger logger = LoggerFactory.getLogger(WebSocketReactiveSocketFactory.class);
+public class WebSocketReactiveSocketConnector implements ReactiveSocketConnector<SocketAddress> {
+    private static final Logger logger = LoggerFactory.getLogger(WebSocketReactiveSocketConnector.class);
 
     private final ConnectionSetupPayload connectionSetupPayload;
     private final Consumer<Throwable> errorStream;
     private final String path;
     private final ClientManager clientManager;
 
-    public WebSocketReactiveSocketFactory(String path, ClientManager clientManager, ConnectionSetupPayload connectionSetupPayload, Consumer<Throwable> errorStream) {
+    public WebSocketReactiveSocketConnector(String path, ClientManager clientManager, ConnectionSetupPayload connectionSetupPayload, Consumer<Throwable> errorStream) {
         this.connectionSetupPayload = connectionSetupPayload;
         this.errorStream = errorStream;
         this.path = path;
         this.clientManager = clientManager;
     }
 
     @Override
-    public Publisher<ReactiveSocket> call(SocketAddress address) {
+    public Publisher<ReactiveSocket> connect(SocketAddress address) {
         Publisher<WebSocketDuplexConnection> connection
             = ReactiveSocketWebSocketClient.create(address, path, clientManager);
 

reactivesocket-local/src/main/java/io/reactivesocket/local/LocalClientDuplexConnection.java

@@ -80,6 +80,11 @@ public void onComplete() {
             });
     }
 
+    @Override
+    public double availability() {
+        return 1.0;
+    }
+
     void write(Frame frame) {
         subjects
             .forEach(o -> o.onNext(frame));

reactivesocket-local/src/main/java/io/reactivesocket/local/LocalClientReactiveSocketFactory.java renamed to reactivesocket-local/src/main/java/io/reactivesocket/local/LocalClientReactiveSocketConnector.java

@@ -15,20 +15,17 @@
  */
 package io.reactivesocket.local;
 
-import io.reactivesocket.ConnectionSetupPayload;
-import io.reactivesocket.DefaultReactiveSocket;
-import io.reactivesocket.ReactiveSocket;
-import io.reactivesocket.ReactiveSocketFactory;
+import io.reactivesocket.*;
 import io.reactivesocket.internal.rx.EmptySubscription;
 import org.reactivestreams.Publisher;
 
-public class LocalClientReactiveSocketFactory implements ReactiveSocketFactory<LocalClientReactiveSocketFactory.Config, ReactiveSocket> {
-    public static final LocalClientReactiveSocketFactory INSTANCE = new LocalClientReactiveSocketFactory();
+public class LocalClientReactiveSocketConnector implements ReactiveSocketConnector<LocalClientReactiveSocketConnector.Config> {
+    public static final LocalClientReactiveSocketConnector INSTANCE = new LocalClientReactiveSocketConnector();
 
-    private LocalClientReactiveSocketFactory() {}
+    private LocalClientReactiveSocketConnector() {}
 
     @Override
-    public Publisher<ReactiveSocket> call(Config config) {
+    public Publisher<ReactiveSocket> connect(Config config) {
         return s -> {
             try {
                 s.onSubscribe(EmptySubscription.INSTANCE);

reactivesocket-local/src/main/java/io/reactivesocket/local/LocalServerDuplexConection.java

@@ -79,6 +79,11 @@ public void onComplete() {
             });
     }
 
+    @Override
+    public double availability() {
+        return 1.0;
+    }
+
     void write(Frame frame) {
         subjects
             .forEach(o -> o.onNext(frame));

reactivesocket-local/src/main/java/io/reactivesocket/local/LocalServerReactiveSocketFactory.java renamed to reactivesocket-local/src/main/java/io/reactivesocket/local/LocalServerReactiveSocketConnector.java

@@ -15,20 +15,17 @@
  */
 package io.reactivesocket.local;
 
-import io.reactivesocket.ConnectionSetupHandler;
-import io.reactivesocket.DefaultReactiveSocket;
-import io.reactivesocket.ReactiveSocket;
-import io.reactivesocket.ReactiveSocketFactory;
+import io.reactivesocket.*;
 import io.reactivesocket.internal.rx.EmptySubscription;
 import org.reactivestreams.Publisher;
 
-public class LocalServerReactiveSocketFactory implements ReactiveSocketFactory<LocalServerReactiveSocketFactory.Config, ReactiveSocket> {
-    public static final LocalServerReactiveSocketFactory INSTANCE = new LocalServerReactiveSocketFactory();
+public class LocalServerReactiveSocketConnector implements ReactiveSocketConnector<LocalServerReactiveSocketConnector.Config> {
+    public static final LocalServerReactiveSocketConnector INSTANCE = new LocalServerReactiveSocketConnector();
 
-    private LocalServerReactiveSocketFactory() {}
+    private LocalServerReactiveSocketConnector() {}
 
     @Override
-    public Publisher<ReactiveSocket> call(Config config) {
+    public Publisher<ReactiveSocket> connect(Config config) {
         return s -> {
             try {
                 s.onSubscribe(EmptySubscription.INSTANCE);

reactivesocket-local/src/test/java/io/reactivesocket/local/ClientServerTest.java

@@ -32,6 +32,8 @@
 
 import java.util.concurrent.TimeUnit;
 
+import static io.reactivesocket.util.Unsafe.toSingleFuture;
+
 public class ClientServerTest {
 
     static ReactiveSocket client;
@@ -40,14 +42,13 @@ public class ClientServerTest {
 
     @BeforeClass
     public static void setup() throws Exception {
-        server = LocalServerReactiveSocketFactory.INSTANCE.callAndWait(new LocalServerReactiveSocketFactory.Config("test", new ConnectionSetupHandler() {
+        LocalServerReactiveSocketConnector.Config serverConfig = new LocalServerReactiveSocketConnector.Config("test", new ConnectionSetupHandler() {
             @Override
             public RequestHandler apply(ConnectionSetupPayload setupPayload, ReactiveSocket rs) throws SetupException {
                 return new RequestHandler() {
                     @Override
                     public Publisher<Payload> handleRequestResponse(Payload payload) {
                         return s -> {
-                            //System.out.println("Handling request/response payload => " + s.toString());
                             Payload response = TestUtil.utf8EncodedPayload("hello world", "metadata");
                             s.onNext(response);
                             s.onComplete();
@@ -91,10 +92,12 @@ public Publisher<Void> handleMetadataPush(Payload payload) {
                     }
                 };
             }
-        }));
+        });
 
-        client = LocalClientReactiveSocketFactory.INSTANCE.callAndWait(new LocalClientReactiveSocketFactory.Config("test", "text", "text"));
+        server = toSingleFuture(LocalServerReactiveSocketConnector.INSTANCE.connect(serverConfig)).get(5, TimeUnit.SECONDS);
 
+        LocalClientReactiveSocketConnector.Config clientConfig = new LocalClientReactiveSocketConnector.Config("test", "text", "text");
+        client = toSingleFuture(LocalClientReactiveSocketConnector.INSTANCE.connect(clientConfig)).get(5, TimeUnit.SECONDS);;
     }
 
     @Test