The connection startup message should only be sent optionally, when cancelling a request it should not be sent.

Author: vietj

vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgConnectionFactory.java

@@ -71,60 +71,61 @@ protected Future<Connection> doConnectInternal(PgConnectOptions options, Context
       return context.failedFuture(e);
     }
     SocketAddress server = options.getSocketAddress();
-    return doConnect(server, context, options);
+    return connect(server, context, true, options);
   }
 
-  public void cancelRequest(PgConnectOptions options, int processId, int secretKey, Handler<AsyncResult<Void>> handler) {
-    doConnect(options.getSocketAddress(), vertx.createEventLoopContext(), options).onComplete(ar -> {
-      if (ar.succeeded()) {
-        PgSocketConnection conn = (PgSocketConnection) ar.result();
-        conn.sendCancelRequestMessage(processId, secretKey, handler);
-      } else {
-        handler.handle(Future.failedFuture(ar.cause()));
-      }
-    });
+  public Future<Void> cancelRequest(PgConnectOptions options, int processId, int secretKey) {
+    return connect(options.getSocketAddress(), vertx.createEventLoopContext(), false, options)
+      .compose(conn -> {
+        PgSocketConnection socket = (PgSocketConnection) conn;
+        return socket.sendCancelRequestMessage(processId, secretKey);
+      });
   }
 
-  private Future<Connection> doConnect(SocketAddress server, ContextInternal context, PgConnectOptions options) {
+  private Future<Connection> connect(SocketAddress server, ContextInternal context, boolean sendStartupMessage, PgConnectOptions options) {
     SslMode sslMode = options.isUsingDomainSocket() ? SslMode.DISABLE : options.getSslMode();
     ConnectOptions connectOptions = new ConnectOptions()
       .setRemoteAddress(server);
     Future<Connection> connFuture;
     switch (sslMode) {
       case DISABLE:
-        connFuture = doConnect(connectOptions, context, false, options);
+        connFuture = connect(connectOptions, context, false, sendStartupMessage, options);
         break;
       case ALLOW:
-        connFuture = doConnect(connectOptions, context, false, options).recover(err -> doConnect(connectOptions, context, true, options));
+        connFuture = connect(connectOptions, context, false, sendStartupMessage, options).recover(err -> connect(connectOptions, context, true, sendStartupMessage, options));
         break;
       case PREFER:
-        connFuture = doConnect(connectOptions, context, true, options).recover(err -> doConnect(connectOptions, context, false, options));
+        connFuture = connect(connectOptions, context, true, sendStartupMessage, options).recover(err -> connect(connectOptions, context, false, sendStartupMessage, options));
         break;
       case REQUIRE:
       case VERIFY_CA:
       case VERIFY_FULL:
-        connFuture = doConnect(connectOptions, context, true, options);
+        connFuture = connect(connectOptions, context, true, sendStartupMessage, options);
         break;
       default:
         return context.failedFuture(new IllegalArgumentException("Unsupported SSL mode"));
     }
     return connFuture;
   }
 
-  private Future<Connection> doConnect(ConnectOptions connectOptions, ContextInternal context, boolean ssl, PgConnectOptions options) {
-    return doConnect_(connectOptions, context, ssl, options).flatMap(conn -> {
-      String username = options.getUser();
-      String password = options.getPassword();
-      String database = options.getDatabase();
-      Map<String, String> properties = options.getProperties() != null ? Collections.unmodifiableMap(options.getProperties()) : null;
-      PgSocketConnection socket = (PgSocketConnection) conn;
-      socket.init();
-      return Future.<Connection>future(p -> socket.sendStartupMessage(username, password, database, properties, p))
-        .map(conn);
-    });
+  private Future<Connection> connect(ConnectOptions connectOptions, ContextInternal context, boolean ssl, boolean sendStartupMessage, PgConnectOptions options) {
+    Future<Connection> res = doConnect(connectOptions, context, ssl, options);
+    if (sendStartupMessage) {
+      return res.flatMap(conn -> {
+        PgSocketConnection socket = (PgSocketConnection) conn;
+        socket.init();
+        String username = options.getUser();
+        String password = options.getPassword();
+        String database = options.getDatabase();
+        Map<String, String> properties = options.getProperties() != null ? Collections.unmodifiableMap(options.getProperties()) : null;
+        return socket.sendStartupMessage(username, password, database, properties);
+      });
+    } else {
+      return res;
+    }
   }
 
-  private Future<Connection> doConnect_(ConnectOptions connectOptions, ContextInternal context, boolean ssl, PgConnectOptions options) {
+  private Future<Connection> doConnect(ConnectOptions connectOptions, ContextInternal context, boolean ssl, PgConnectOptions options) {
     Future<NetSocket> soFut;
     try {
       soFut = client.connect(connectOptions);

vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgConnectionImpl.java

@@ -123,7 +123,7 @@ public Future<Void> cancelRequest() {
     Promise<Void> promise = context.owner().getOrCreateContext().promise();
     context.emit(promise, p -> {
       PgSocketConnection unwrap = (PgSocketConnection) conn.unwrap();
-      ((PgConnectionFactory) factory).cancelRequest(unwrap.connectOptions(), this.processId(), this.secretKey(), p);
+      ((PgConnectionFactory) factory).cancelRequest(unwrap.connectOptions(), this.processId(), this.secretKey()).onComplete(p);
     });
     return promise.future();
   }

vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgSocketConnection.java

@@ -83,29 +83,26 @@ public void init() {
   }
 
   // TODO RETURN FUTURE ???
-  void sendStartupMessage(String username, String password, String database, Map<String, String> properties, Promise<Connection> completionHandler) {
+  Future<Connection> sendStartupMessage(String username, String password, String database, Map<String, String> properties) {
     InitCommand cmd = new InitCommand(this, username, password, database, properties);
-    schedule(context, cmd).onComplete(completionHandler);
+    return schedule(context, cmd);
   }
 
-  void sendCancelRequestMessage(int processId, int secretKey, Handler<AsyncResult<Void>> handler) {
+  Future<Void> sendCancelRequestMessage(int processId, int secretKey) {
     Buffer buffer = Buffer.buffer(16);
     buffer.appendInt(16);
     // cancel request code
     buffer.appendInt(80877102);
     buffer.appendInt(processId);
     buffer.appendInt(secretKey);
 
-    socket.write(buffer).onComplete(ar -> {
+    return socket.write(buffer).andThen(ar -> {
       if (ar.succeeded()) {
         // directly close this connection
         if (status == Status.CONNECTED) {
           status = Status.CLOSING;
           socket.close();
         }
-        handler.handle(Future.succeededFuture());
-      } else {
-        handler.handle(Future.failedFuture(ar.cause()));
       }
     });
   }