Introduce Connector and availability for Connection

Problem
The DuplexConnection class doesn't expose any way to probe its state,
this is somewhat problematic and doesn't fit well in the current
model where the availability is the composable way of chaining state.

The ReactiveSocketFactory create a ReactiveSocket from an `address`
(currently SocketAddress). There are places in the code where the concept
of address is not needed and it leads to implicit dependency on it.

Solution
Add a `double availability()` method to the DuplexConnection interface,
most implementation of this method will be straightforward (just
returning 0.0 if the underlying resource is unavailable, 1.0 otherwise).

Split the concept of ReactiveSocketFactory in two, the Factory and the
Connector. The Connector is responsible for creating the ReactiveSocket
from the address, and the factory can create a ReactiveSocket without
any argument.

Author: stevegury

src/main/java/io/reactivesocket/DefaultReactiveSocket.java

@@ -440,6 +440,10 @@ public void addOutput(Frame f, Completable callback) {
             connection.addOutput(f, callback);
         }
 
+        @Override
+        public double availability() {
+            return connection.availability();
+        }
     };
 
     @Override

src/main/java/io/reactivesocket/DuplexConnection.java

@@ -38,4 +38,10 @@ default void addOutput(Frame frame, Completable callback) {
             s.onComplete();
         }, callback);
 	}
+
+    /**
+     * @return the availability of the underlying connection, a number in [0.0, 1.0]
+     * (higher is better).
+     */
+    double availability();
 }

src/main/java/io/reactivesocket/ReactiveSocket.java

@@ -26,10 +26,10 @@
  * Interface for a connection that supports sending requests and receiving responses
  */
 public interface ReactiveSocket extends AutoCloseable {
-    Publisher<Payload> requestResponse(final Payload payload);
-
     Publisher<Void> fireAndForget(final Payload payload);
 
+    Publisher<Payload> requestResponse(final Payload payload);
+
     Publisher<Payload> requestStream(final Payload payload);
 
     Publisher<Payload> requestSubscription(final Payload payload);

src/main/java/io/reactivesocket/ReactiveSocketFactory.java

@@ -15,102 +15,29 @@
  */
 package io.reactivesocket;
 
-import io.reactivesocket.internal.rx.EmptySubscription;
 import org.reactivestreams.Publisher;
-import org.reactivestreams.Subscriber;
-import org.reactivestreams.Subscription;
-
-import java.util.NoSuchElementException;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-@FunctionalInterface
-public interface ReactiveSocketFactory<T, R extends ReactiveSocket> {
-
-    Publisher<R> call(T t);
+import java.net.SocketAddress;
 
+/**
+ * Factory of ReactiveSocket interface
+ * This abstraction is useful for abstracting the creation of a ReactiveSocket
+ * (e.g. inside the LoadBalancer which create ReactiveSocket as needed)
+ */
+public interface ReactiveSocketFactory<T> {
     /**
-     * Gets a socket in a blocking manner
-     * @param t configuration to create the reactive socket
-     * @return blocks on create the socket
+     * Construct the ReactiveSocket
+     * @return
      */
-    default R callAndWait(T t) {
-        CompletableFuture<R> future = new CompletableFuture<>();
-
-        call(t)
-            .subscribe(new Subscriber<R>() {
-                @Override
-                public void onSubscribe(Subscription s) {
-                    s.request(1);
-                }
-
-                @Override
-                public void onNext(R reactiveSocket) {
-                    future.complete(reactiveSocket);
-                }
-
-                @Override
-                public void onError(Throwable t) {
-                    future.completeExceptionally(t);
-                }
-
-                @Override
-                public void onComplete() {
-                    future.completeExceptionally(new NoSuchElementException("Sequence contains no elements"));
-                }
-            });
-
-        return future.join();
-    }
+    Publisher<ReactiveSocket> apply();
 
     /**
-     *
-     * @param t the configuration used to create the reactive socket
-     * @param timeout timeout
-     * @param timeUnit timeout units
-     * @param executorService ScheduledExecutorService to schedule the timeout on
-     * @return
+     * @return a positive numbers representing the availability of the factory.
+     * Higher is better, 0.0 means not available
      */
-    default Publisher<R> call(T t, long timeout, TimeUnit timeUnit, ScheduledExecutorService executorService) {
-        Publisher<R> reactiveSocketPublisher = subscriber -> {
-            AtomicBoolean complete = new AtomicBoolean();
-            subscriber.onSubscribe(EmptySubscription.INSTANCE);
-            call(t)
-                .subscribe(new Subscriber<R>() {
-                    @Override
-                    public void onSubscribe(Subscription s) {
-                        s.request(1);
-                    }
-
-                    @Override
-                    public void onNext(R reactiveSocket) {
-                        subscriber.onNext(reactiveSocket);
-                    }
-
-                    @Override
-                    public void onError(Throwable t) {
-                        subscriber.onError(t);
-                    }
-
-                    @Override
-                    public void onComplete() {
-                        if (complete.compareAndSet(false, true)) {
-                            subscriber.onComplete();
-                        }
-                    }
-                });
-
-            executorService.schedule(() -> {
-                if (complete.compareAndSet(false, true)) {
-                    subscriber.onError(new TimeoutException());
-                }
-            }, timeout, timeUnit);
-        };
-
-        return reactiveSocketPublisher;
-    }
+    double availability();
 
+    /**
+     * @return an identifier of the remote location
+     */
+    T remote();
 }

src/main/java/io/reactivesocket/internal/Requester.java

@@ -286,14 +286,14 @@ private void assertStarted() {
      */
     public double availability() {
         if (!honorLease) {
-            return 1.0;
+            return connection.availability();
         }
         final long now = System.currentTimeMillis();
         double available = 0.0;
         if (numberOfRemainingRequests > 0 && (now < ttlExpiration)) {
             available = 1.0;
         }
-        return available;
+        return available * connection.availability();
     }
 
     /*
@@ -873,6 +873,7 @@ public void error(Throwable e) {
 
                         Publisher<Frame> keepaliveTicker =
                             PublisherUtils.keepaliveTicker(KEEPALIVE_INTERVAL_MS, TimeUnit.MILLISECONDS);
+
                         connection.addOutput(keepaliveTicker,
                             new Completable() {
                                 public void success() {}

src/perf/java/io/reactivesocket/perfutil/PerfTestConnection.java

@@ -64,6 +64,11 @@ public void addOutput(Frame f, Completable callback) {
 		callback.success();
 	}
 
+	@Override
+	public double availability() {
+		return 1.0;
+	}
+
 	@Override
 	public Observable<Frame> getInput() {
 		return toInput;
@@ -72,11 +77,9 @@ public Observable<Frame> getInput() {
 	public void connectToServerConnection(PerfTestConnection serverConnection) {
 		writeSubject.subscribe(serverConnection.toInput);
 		serverConnection.writeSubject.subscribe(toInput);
-
 	}
 
 	@Override
 	public void close() throws IOException {
-		
 	}
 }

src/test/java/io/reactivesocket/TestConnection.java

@@ -48,6 +48,11 @@ public void addOutput(Frame f, Completable callback) {
         callback.success();
 	}
 
+	@Override
+	public double availability() {
+		return 1.0;
+	}
+
 	@Override
 	public io.reactivesocket.rx.Observable<Frame> getInput() {
 		return new io.reactivesocket.rx.Observable<Frame>() {