Propagate SSL failures for SSL disabled/required

If a client requires SSL but the server has SSL disabled, the failure is
now correctly propagated to the client.

Previously, the connection was hanging.

[resolves #187]

Author: mp911de

src/main/java/io/r2dbc/postgresql/client/SSLSessionHandlerAdapter.java

@@ -25,6 +25,7 @@
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.GenericFutureListener;
 import io.r2dbc.postgresql.message.frontend.SSLRequest;
+import io.r2dbc.spi.R2dbcPermissionDeniedException;
 import reactor.core.publisher.Mono;
 
 import javax.net.ssl.SSLEngine;
@@ -87,21 +88,26 @@ public void operationComplete(Future<Channel> future) throws Exception {
         if (this.sslConfig.getHostnameVerifier().verify(hostName, this.sslEngine.getSession())) {
             this.handshakeFuture.complete(null);
         } else {
-            this.handshakeFuture.completeExceptionally(new IllegalStateException(String.format("The hostname '%s' could not be verified.", socketAddress.getAddress().toString())));
+            this.handshakeFuture.completeExceptionally(new PostgresqlSslException(String.format("The hostname '%s' could not be verified.", socketAddress.getAddress().toString())));
         }
     }
 
     private void processSslDisabled(ChannelHandlerContext ctx, Object msg) {
         if (this.sslConfig.getSslMode().requireSsl()) {
-            throw new IllegalStateException("Server support for SSL connection is disabled, but client was configured with SSL mode " + this.sslConfig.getSslMode());
+            PostgresqlSslException e =
+                new PostgresqlSslException("Server support for SSL connection is disabled, but client was configured with SSL mode " + this.sslConfig.getSslMode());
+            this.handshakeFuture.completeExceptionally(e);
         } else {
             this.handshakeFuture.complete(null);
         }
     }
 
     private void processSslEnabled(ChannelHandlerContext ctx, Object msg) {
         if (this.sslConfig.getSslMode() == SSLMode.DISABLE) {
-            throw new IllegalStateException("Server requires SSL handshake, but client was configured with SSL mode DISABLE");
+
+            PostgresqlSslException e = new PostgresqlSslException("Server requires SSL handshake, but client was configured with SSL mode DISABLE");
+            this.handshakeFuture.completeExceptionally(e);
+            return;
         }
         ctx.channel().pipeline()
             .addFirst(this.sslHandler)
@@ -117,4 +123,15 @@ public void handlerAdded(ChannelHandlerContext ctx) {
     Mono<Void> getHandshake() {
         return Mono.fromFuture(this.handshakeFuture);
     }
+
+    /**
+     * Postgres-specific {@link R2dbcPermissionDeniedException}.
+     */
+    static final class PostgresqlSslException extends R2dbcPermissionDeniedException {
+
+
+        PostgresqlSslException(String msg) {
+            super(msg);
+        }
+    }
 }

src/test/java/io/r2dbc/postgresql/client/ReactorNettyClientIntegrationTests.java

@@ -161,12 +161,13 @@ void constructorNoHost() {
 
     @BeforeEach
     void createTable() {
+        dropTable();
         SERVER.getJdbcOperations().execute("CREATE TABLE test ( value INTEGER )");
     }
 
     @AfterEach
     void dropTable() {
-        SERVER.getJdbcOperations().execute("DROP TABLE test");
+        SERVER.getJdbcOperations().execute("DROP TABLE IF EXISTS test");
     }
 
     @Test
@@ -396,7 +397,7 @@ void invalidServerCertificate() {
                     .sslRootCert(SERVER.getClientCrt()),
                 c -> c
                     .as(StepVerifier::create)
-                    .expectError()
+                    .expectError(R2dbcNonTransientResourceException.class)
                     .verify());
         }
 
@@ -664,7 +665,7 @@ void verifyFullFailedWithWrongHost() {
                         .sslMode(SSLMode.VERIFY_FULL),
                     c -> c
                         .as(StepVerifier::create)
-                        .verifyError(R2dbcNonTransientResourceException.class));
+                        .verifyError(R2dbcPermissionDeniedException.class));
             }
 
             @Test