Correctly wire subscription from Transport Connectors. (#97)

**Problem**
The initialization of the Publisher chain is usually started from
a `onSubscribe` call which was lacking in the transport implementations
we care about.

**Solution**
Add a call to `onSubscribe` with an EmptySubscriber because there's no
notion of back-pressure here, we only eagerly deliver one `ReactiveSocket`.

Author: stevegury

reactivesocket-transport-tcp/src/main/java/io/reactivesocket/transport/tcp/client/ClientTcpDuplexConnection.java

@@ -25,6 +25,7 @@
 import io.reactivesocket.DuplexConnection;
 import io.reactivesocket.Frame;
 import io.reactivesocket.exceptions.TransportException;
+import io.reactivesocket.internal.rx.EmptySubscription;
 import io.reactivesocket.rx.Completable;
 import io.reactivesocket.rx.Observable;
 import io.reactivesocket.rx.Observer;
@@ -73,6 +74,7 @@ protected void initChannel(SocketChannel ch) throws Exception {
             connect.addListener(connectFuture -> {
                 if (connectFuture.isSuccess()) {
                     Channel ch = connect.channel();
+                    s.onSubscribe(EmptySubscription.INSTANCE);
                     s.onNext(new ClientTcpDuplexConnection(ch, subjects));
                     s.onComplete();
                 } else {

reactivesocket-transport-tcp/src/main/java/io/reactivesocket/transport/tcp/client/TcpReactiveSocketConnector.java

@@ -17,6 +17,7 @@
 
 import io.netty.channel.EventLoopGroup;
 import io.reactivesocket.*;
+import io.reactivesocket.internal.rx.EmptySubscription;
 import io.reactivesocket.rx.Completable;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
@@ -57,6 +58,7 @@ public void onNext(ClientTcpDuplexConnection connection) {
                 reactiveSocket.start(new Completable() {
                     @Override
                     public void success() {
+                        s.onSubscribe(EmptySubscription.INSTANCE);
                         s.onNext(reactiveSocket);
                         s.onComplete();
                     }

reactivesocket-transport-websocket/src/main/java/io/reactivesocket/transport/websocket/client/ClientWebSocketDuplexConnection.java

@@ -28,6 +28,7 @@
 import io.reactivesocket.DuplexConnection;
 import io.reactivesocket.Frame;
 import io.reactivesocket.exceptions.TransportException;
+import io.reactivesocket.internal.rx.EmptySubscription;
 import io.reactivesocket.rx.Completable;
 import io.reactivesocket.rx.Observable;
 import io.reactivesocket.rx.Observer;
@@ -91,6 +92,7 @@ protected void initChannel(SocketChannel ch) throws Exception {
                         .getHandshakePromise()
                         .addListener(handshakeFuture -> {
                             if (handshakeFuture.isSuccess()) {
+                                s.onSubscribe(EmptySubscription.INSTANCE);
                                 s.onNext(new ClientWebSocketDuplexConnection(ch, subjects));
                                 s.onComplete();
                             } else {

reactivesocket-transport-websocket/src/main/java/io/reactivesocket/transport/websocket/client/WebSocketReactiveSocketConnector.java

@@ -17,14 +17,13 @@
 
 import io.netty.channel.EventLoopGroup;
 import io.reactivesocket.*;
+import io.reactivesocket.internal.rx.EmptySubscription;
 import io.reactivesocket.rx.Completable;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import rx.Observable;
-import rx.RxReactiveStreams;
 
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
@@ -54,42 +53,40 @@ public Publisher<ReactiveSocket> connect(SocketAddress address) {
             Publisher<ClientWebSocketDuplexConnection> connection
                     = ClientWebSocketDuplexConnection.create((InetSocketAddress)address, path, eventLoopGroup);
 
-            Observable<ReactiveSocket> result = Observable.create(s ->
-                connection.subscribe(new Subscriber<ClientWebSocketDuplexConnection>() {
-                    @Override
-                    public void onSubscribe(Subscription s) {
-                        s.request(1);
-                    }
+            return s -> connection.subscribe(new Subscriber<ClientWebSocketDuplexConnection>() {
+                @Override
+                public void onSubscribe(Subscription s) {
+                    s.request(1);
+                }
 
-                    @Override
-                    public void onNext(ClientWebSocketDuplexConnection connection) {
-                        ReactiveSocket reactiveSocket = DefaultReactiveSocket.fromClientConnection(connection, connectionSetupPayload, errorStream);
-                        reactiveSocket.start(new Completable() {
-                            @Override
-                            public void success() {
-                                s.onNext(reactiveSocket);
-                                s.onCompleted();
-                            }
+                @Override
+                public void onNext(ClientWebSocketDuplexConnection connection) {
+                    ReactiveSocket reactiveSocket = DefaultReactiveSocket.fromClientConnection(connection, connectionSetupPayload, errorStream);
+                    reactiveSocket.start(new Completable() {
+                        @Override
+                        public void success() {
+                            s.onSubscribe(EmptySubscription.INSTANCE);
+                            s.onNext(reactiveSocket);
+                            s.onComplete();
+                        }
 
-                            @Override
-                            public void error(Throwable e) {
-                                s.onError(e);
-                            }
-                        });
-                    }
+                        @Override
+                        public void error(Throwable e) {
+                            s.onError(e);
+                        }
+                    });
+                }
 
-                    @Override
-                    public void onError(Throwable t) {
-                        s.onError(t);
-                    }
+                @Override
+                public void onError(Throwable t) {
+                    s.onError(t);
+                }
 
-                    @Override
-                    public void onComplete() {
-                    }
-                })
-            );
-
-            return RxReactiveStreams.toPublisher(result);
+                @Override
+                public void onComplete() {
+                    s.onComplete();
+                }
+            });
         } else {
             throw new IllegalArgumentException("unknown socket address type => " + address.getClass());
         }