Wrap IOException in ServiceUnavailableException

When establishing connection it was possible for `IOException` to get
propagated directly to the user without being wrapped in a
`ServiceUnavailableException`. This happened because `HandshakeHandler`
only wrapped security-related exceptions with driver exceptions. It was
problematic because rest of the codebase expects to deal with
`ServiceUnavailableException`s. For example rediscovery and retries
code can deal nicely with `ServiceUnavailableException`.

This commit fixes the problem by making `HandshakeHandler` wrap all
non-security related exceptions in `ServiceUnavailableException`.

Author: lutovich

driver/src/main/java/org/neo4j/driver/internal/async/HandshakeHandler.java

@@ -95,16 +95,8 @@ public void exceptionCaught( ChannelHandlerContext ctx, Throwable error )
         else
         {
             failed = true;
-
-            Throwable cause = error instanceof DecoderException ? error.getCause() : error;
-            if ( cause instanceof SSLHandshakeException )
-            {
-                fail( ctx, new SecurityException( "Failed to establish secured connection with the server", cause ) );
-            }
-            else
-            {
-                fail( ctx, cause );
-            }
+            Throwable cause = transformError( error );
+            fail( ctx, cause );
         }
     }
 
@@ -161,4 +153,21 @@ private static Throwable protocolNoSupportedByDriverError( int suggestedProtocol
         return new ClientException(
                 "Protocol error, server suggested unexpected protocol version: " + suggestedProtocolVersion );
     }
+
+    private static Throwable transformError( Throwable error )
+    {
+        Throwable cause = error instanceof DecoderException ? error.getCause() : error;
+        if ( cause instanceof ServiceUnavailableException )
+        {
+            return cause;
+        }
+        else if ( cause instanceof SSLHandshakeException )
+        {
+            return new SecurityException( "Failed to establish secured connection with the server", cause );
+        }
+        else
+        {
+            return new ServiceUnavailableException( "Failed to establish connection with the server", cause );
+        }
+    }
 }

driver/src/main/java/org/neo4j/driver/internal/async/inbound/ChannelErrorHandler.java

@@ -101,7 +101,7 @@ private void fail( ChannelHandlerContext ctx, Throwable error )
         ctx.close();
     }
 
-    private Throwable transformError( Throwable error )
+    private static Throwable transformError( Throwable error )
     {
         if ( error instanceof CodecException )
         {

driver/src/test/java/org/neo4j/driver/internal/async/ChannelConnectorImplIT.java

@@ -29,7 +29,9 @@
 import org.junit.Test;
 
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.net.ServerSocket;
+import java.net.Socket;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 
@@ -44,6 +46,7 @@
 import org.neo4j.driver.v1.exceptions.ServiceUnavailableException;
 import org.neo4j.driver.v1.util.TestNeo4j;
 
+import static java.util.concurrent.CompletableFuture.runAsync;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.startsWith;
 import static org.junit.Assert.assertEquals;
@@ -186,6 +189,43 @@ public void shouldFailWhenTLSHandshakeTakesTooLong() throws Exception
         testReadTimeoutOnConnect( SecurityPlan.forAllCertificates() );
     }
 
+    @Test
+    public void shouldThrowServiceUnavailableExceptionOnFailureDuringConnect() throws Exception
+    {
+        ServerSocket server = new ServerSocket( 0 );
+        BoltServerAddress address = new BoltServerAddress( "localhost", server.getLocalPort() );
+
+        runAsync( () ->
+        {
+            try
+            {
+                // wait for a connection
+                Socket socket = server.accept();
+                // and terminate it immediately so that client gets a "reset by peer" IOException
+                socket.close();
+                server.close();
+            }
+            catch ( IOException e )
+            {
+                throw new UncheckedIOException( e );
+            }
+        } );
+
+        ChannelConnector connector = newConnector( neo4j.authToken() );
+        ChannelFuture channelFuture = connector.connect( address, bootstrap );
+
+        try
+        {
+            await( channelFuture );
+            fail( "Exception expected" );
+        }
+        catch ( ServiceUnavailableException e )
+        {
+            assertEquals( "Failed to establish connection with the server", e.getMessage() );
+            assertThat( e.getCause(), instanceOf( IOException.class ) );
+        }
+    }
+
     private void testReadTimeoutOnConnect( SecurityPlan securityPlan ) throws IOException
     {
         try ( ServerSocket server = new ServerSocket( 0 ) ) // server that accepts connections but does not reply

driver/src/test/java/org/neo4j/driver/internal/async/HandshakeHandlerTest.java

@@ -46,10 +46,10 @@
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.fail;
-import static org.neo4j.driver.internal.async.ChannelAttributes.setMessageDispatcher;
 import static org.neo4j.driver.internal.async.BoltProtocolV1Util.HTTP;
 import static org.neo4j.driver.internal.async.BoltProtocolV1Util.NO_PROTOCOL_VERSION;
 import static org.neo4j.driver.internal.async.BoltProtocolV1Util.PROTOCOL_VERSION_1;
+import static org.neo4j.driver.internal.async.ChannelAttributes.setMessageDispatcher;
 import static org.neo4j.driver.internal.logging.DevNullLogging.DEV_NULL_LOGGING;
 import static org.neo4j.driver.v1.util.TestUtil.await;
 
@@ -85,9 +85,34 @@ public void shouldFailGivenPromiseWhenExceptionCaught()
             await( handshakeCompletedPromise );
             fail( "Exception expected" );
         }
-        catch ( Exception e )
+        catch ( ServiceUnavailableException e )
+        {
+            assertEquals( cause, e.getCause() );
+        }
+
+        // channel should be closed
+        assertNull( await( channel.closeFuture() ) );
+    }
+
+    @Test
+    public void shouldFailGivenPromiseWhenServiceUnavailableExceptionCaught()
+    {
+        ChannelPromise handshakeCompletedPromise = channel.newPromise();
+        HandshakeHandler handler = newHandler( handshakeCompletedPromise );
+        channel.pipeline().addLast( handler );
+
+        ServiceUnavailableException error = new ServiceUnavailableException( "Bad error" );
+        channel.pipeline().fireExceptionCaught( error );
+
+        try
         {
-            assertEquals( cause, e );
+            // promise should fail
+            await( handshakeCompletedPromise );
+            fail( "Exception expected" );
+        }
+        catch ( ServiceUnavailableException e )
+        {
+            assertEquals( error, e );
         }
 
         // channel should be closed
@@ -112,9 +137,9 @@ public void shouldFailGivenPromiseWhenMultipleExceptionsCaught()
             await( handshakeCompletedPromise );
             fail( "Exception expected" );
         }
-        catch ( RuntimeException e )
+        catch ( ServiceUnavailableException e )
         {
-            assertEquals( error1, e );
+            assertEquals( error1, e.getCause() );
         }
 
         // channel should be closed
@@ -147,9 +172,9 @@ public void shouldUnwrapDecoderException()
             await( handshakeCompletedPromise );
             fail( "Exception expected" );
         }
-        catch ( Exception e )
+        catch ( ServiceUnavailableException e )
         {
-            assertEquals( cause, e );
+            assertEquals( cause, e.getCause() );
         }
 
         // channel should be closed