ClientChannelFactoryImpl was not catching errors inside operationComplete()

This caused any errors not getting propagated to the connect() subscriber.

Also added some debug for a flaky test.

Author: Nitesh Kant

rx-netty/src/main/java/io/reactivex/netty/client/ClientChannelFactoryImpl.java

@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package io.reactivex.netty.client;
 
 import io.netty.bootstrap.Bootstrap;
@@ -74,34 +75,38 @@ public void call() {
         connectFuture.addListener(new ChannelFutureListener() {
             @Override
             public void operationComplete(ChannelFuture future) throws Exception {
-                if (!future.isSuccess()) {
-                    eventsSubject.onEvent(ClientMetricsEvent.CONNECT_FAILED, Clock.onEndMillis(startTimeMillis),
-                                          future.cause());
-                    subscriber.onError(future.cause());
-                } else {
-                    eventsSubject.onEvent(ClientMetricsEvent.CONNECT_SUCCESS, Clock.onEndMillis(startTimeMillis));
-                    ChannelPipeline pipeline = future.channel().pipeline();
-                    ChannelHandlerContext ctx = pipeline.lastContext(); // The connection uses the context for write which should always start from the tail.
-                    final ObservableConnection<I, O> newConnection = connectionFactory.newConnection(ctx);
-                    ChannelHandler lifecycleHandler = pipeline.get(RxRequiredConfigurator.CONN_LIFECYCLE_HANDLER_NAME);
-                    if (null == lifecycleHandler) {
-                        onNewConnection(newConnection, subscriber);
+                try {
+                    if (!future.isSuccess()) {
+                        eventsSubject.onEvent(ClientMetricsEvent.CONNECT_FAILED, Clock.onEndMillis(startTimeMillis),
+                                              future.cause());
+                        subscriber.onError(future.cause());
                     } else {
-                        @SuppressWarnings("unchecked")
-                        ConnectionLifecycleHandler<I, O> handler = (ConnectionLifecycleHandler<I, O>) lifecycleHandler;
-                        SslHandler sslHandler = pipeline.get(SslHandler.class);
-                        if (null == sslHandler) {
-                            handler.setConnection(newConnection);
+                        eventsSubject.onEvent(ClientMetricsEvent.CONNECT_SUCCESS, Clock.onEndMillis(startTimeMillis));
+                        ChannelPipeline pipeline = future.channel().pipeline();
+                        ChannelHandlerContext ctx = pipeline.lastContext(); // The connection uses the context for write which should always start from the tail.
+                        final ObservableConnection<I, O> newConnection = connectionFactory.newConnection(ctx);
+                        ChannelHandler lifecycleHandler = pipeline.get(RxRequiredConfigurator.CONN_LIFECYCLE_HANDLER_NAME);
+                        if (null == lifecycleHandler) {
                             onNewConnection(newConnection, subscriber);
                         } else {
-                            sslHandler.handshakeFuture().addListener(new GenericFutureListener<Future<? super Channel>>() {
-                                @Override
-                                public void operationComplete(Future<? super Channel> future) throws Exception {
-                                    onNewConnection(newConnection, subscriber);
-                                }
-                            });
+                            @SuppressWarnings("unchecked")
+                            ConnectionLifecycleHandler<I, O> handler = (ConnectionLifecycleHandler<I, O>) lifecycleHandler;
+                            SslHandler sslHandler = pipeline.get(SslHandler.class);
+                            if (null == sslHandler) {
+                                handler.setConnection(newConnection);
+                                onNewConnection(newConnection, subscriber);
+                            } else {
+                                sslHandler.handshakeFuture().addListener(new GenericFutureListener<Future<? super Channel>>() {
+                                    @Override
+                                    public void operationComplete(Future<? super Channel> future) throws Exception {
+                                        onNewConnection(newConnection, subscriber);
+                                    }
+                                });
+                            }
                         }
                     }
+                } catch (Throwable throwable) {
+                    subscriber.onError(throwable);
                 }
             }
         });

rx-netty/src/main/java/io/reactivex/netty/server/AbstractServer.java

@@ -28,6 +28,8 @@
 import io.reactivex.netty.metrics.MetricEventsSubject;
 import io.reactivex.netty.pipeline.PipelineConfigurator;
 import io.reactivex.netty.pipeline.PipelineConfiguratorComposite;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import rx.Subscription;
 
 import java.net.InetSocketAddress;
@@ -42,6 +44,8 @@
 public class AbstractServer<I, O, B extends AbstractBootstrap<B, C>, C extends Channel, S extends AbstractServer>
         implements MetricEventsPublisher<ServerMetricsEvent<?>> {
 
+    private static final Logger logger = LoggerFactory.getLogger(AbstractServer.class);
+
     protected enum ServerState {Created, Starting, Started, Shutdown}
 
     protected final UnpooledConnectionFactory<I,O> connectionFactory;
@@ -88,6 +92,8 @@ public S start() {
 
         serverStateRef.set(ServerState.Started); // It will come here only if this was the thread that transitioned to Starting
 
+        logger.info("Rx server started at port: " + getServerPort());
+
         return returnServer();
     }
 

rx-netty/src/test/java/io/reactivex/netty/protocol/http/server/UnexpectedErrorsTest.java

@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package io.reactivex.netty.protocol.http.server;
 
 import io.netty.buffer.ByteBuf;
@@ -72,7 +73,7 @@ public void tearDown() throws Exception {
     public void testErrorHandlerReturnsNull() throws Exception {
         TestableErrorHandler errorHandler = new TestableErrorHandler(null);
         server.withErrorHandler(errorHandler).start();
-
+        System.err.println("[testErrorHandlerReturnsNull] Server port: " + server.getServerPort());
         blockTillConnected(server.getServerPort());
         channelCloseListener.waitForClose(1, TimeUnit.MINUTES);
 
@@ -86,6 +87,8 @@ public void testConnectionHandlerReturnsError() throws Exception {
 
         server.withErrorHandler(errorHandler).start();
 
+        System.err.println("[testConnectionHandlerReturnsError] Server port: " + server.getServerPort());
+
         blockTillConnected(server.getServerPort());
 
         channelCloseListener.waitForClose(1, TimeUnit.MINUTES);