Merge SqlConnectionBase and SqlConnectionImpl in a single class

vietj · vietj · commit 65f7a8d8ea4d · 2021-12-08T12:01:49.000+01:00
diff --git a/vertx-db2-client/src/main/java/io/vertx/db2client/impl/DB2ConnectionImpl.java b/vertx-db2-client/src/main/java/io/vertx/db2client/impl/DB2ConnectionImpl.java
@@ -26,11 +26,11 @@
 import io.vertx.db2client.impl.command.PingCommand;
 import io.vertx.db2client.spi.DB2Driver;
 import io.vertx.sqlclient.impl.Connection;
-import io.vertx.sqlclient.impl.SqlConnectionImpl;
+import io.vertx.sqlclient.impl.SqlConnectionBase;
 import io.vertx.sqlclient.impl.tracing.QueryTracer;
 import io.vertx.sqlclient.spi.ConnectionFactory;
 
-public class DB2ConnectionImpl extends SqlConnectionImpl<DB2ConnectionImpl> implements DB2Connection {
+public class DB2ConnectionImpl extends SqlConnectionBase<DB2ConnectionImpl> implements DB2Connection {
 
   public static Future<DB2Connection> connect(Vertx vertx, DB2ConnectOptions options) {
     ContextInternal ctx = (ContextInternal) vertx.getOrCreateContext();
diff --git a/vertx-db2-client/src/main/java/io/vertx/db2client/impl/DB2PoolImpl.java b/vertx-db2-client/src/main/java/io/vertx/db2client/impl/DB2PoolImpl.java
@@ -29,7 +29,6 @@
 import io.vertx.sqlclient.SqlConnection;
 import io.vertx.sqlclient.impl.Connection;
 import io.vertx.sqlclient.impl.PoolBase;
-import io.vertx.sqlclient.impl.SqlConnectionImpl;
 import io.vertx.sqlclient.impl.tracing.QueryTracer;
 import io.vertx.sqlclient.spi.ConnectionFactory;
 
@@ -43,7 +42,7 @@ public DB2PoolImpl(VertxInternal vertx, int pipeliningLimit, PoolOptions poolOpt
 
   @SuppressWarnings("rawtypes")
   @Override
-  protected SqlConnectionImpl wrap(ContextInternal context, ConnectionFactory factory, Connection conn) {
+  protected DB2ConnectionImpl wrap(ContextInternal context, ConnectionFactory factory, Connection conn) {
     return new DB2ConnectionImpl(context, factory, conn, tracer, metrics);
   }
 
diff --git a/vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/MSSQLConnectionImpl.java b/vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/MSSQLConnectionImpl.java
@@ -19,11 +19,11 @@
 import io.vertx.core.Vertx;
 import io.vertx.mssqlclient.spi.MSSQLDriver;
 import io.vertx.sqlclient.impl.Connection;
-import io.vertx.sqlclient.impl.SqlConnectionImpl;
+import io.vertx.sqlclient.impl.SqlConnectionBase;
 import io.vertx.sqlclient.impl.tracing.QueryTracer;
 import io.vertx.sqlclient.spi.ConnectionFactory;
 
-public class MSSQLConnectionImpl extends SqlConnectionImpl<MSSQLConnectionImpl> implements MSSQLConnection {
+public class MSSQLConnectionImpl extends SqlConnectionBase<MSSQLConnectionImpl> implements MSSQLConnection {
 
   public MSSQLConnectionImpl(ContextInternal context, ConnectionFactory factory, Connection conn, QueryTracer tracer, ClientMetrics metrics) {
     super(context, factory, conn, MSSQLDriver.INSTANCE, tracer, metrics);
diff --git a/vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/MSSQLPoolImpl.java b/vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/MSSQLPoolImpl.java
@@ -25,7 +25,6 @@
 import io.vertx.sqlclient.SqlConnection;
 import io.vertx.sqlclient.impl.Connection;
 import io.vertx.sqlclient.impl.PoolBase;
-import io.vertx.sqlclient.impl.SqlConnectionImpl;
 import io.vertx.sqlclient.impl.tracing.QueryTracer;
 import io.vertx.sqlclient.spi.ConnectionFactory;
 
@@ -38,7 +37,7 @@ public MSSQLPoolImpl(VertxInternal vertx, MSSQLConnectOptions baseConnectOptions
   }
 
   @Override
-  protected SqlConnectionImpl wrap(ContextInternal context, ConnectionFactory factory, Connection connection) {
+  protected MSSQLConnectionImpl wrap(ContextInternal context, ConnectionFactory factory, Connection connection) {
     return new MSSQLConnectionImpl(context, factory, connection, tracer, metrics);
   }
 
diff --git a/vertx-mysql-client/src/main/java/io/vertx/mysqlclient/impl/MySQLConnectionImpl.java b/vertx-mysql-client/src/main/java/io/vertx/mysqlclient/impl/MySQLConnectionImpl.java
@@ -24,11 +24,11 @@
 import io.vertx.mysqlclient.impl.command.*;
 import io.vertx.mysqlclient.spi.MySQLDriver;
 import io.vertx.sqlclient.impl.Connection;
-import io.vertx.sqlclient.impl.SqlConnectionImpl;
+import io.vertx.sqlclient.impl.SqlConnectionBase;
 import io.vertx.sqlclient.impl.tracing.QueryTracer;
 import io.vertx.sqlclient.spi.ConnectionFactory;
 
-public class MySQLConnectionImpl extends SqlConnectionImpl<MySQLConnectionImpl> implements MySQLConnection {
+public class MySQLConnectionImpl extends SqlConnectionBase<MySQLConnectionImpl> implements MySQLConnection {
 
   public static Future<MySQLConnection> connect(ContextInternal ctx, MySQLConnectOptions options) {
     if (options.isUsingDomainSocket() && !ctx.owner().isNativeTransportEnabled()) {
diff --git a/vertx-mysql-client/src/main/java/io/vertx/mysqlclient/impl/MySQLPoolImpl.java b/vertx-mysql-client/src/main/java/io/vertx/mysqlclient/impl/MySQLPoolImpl.java
@@ -25,7 +25,6 @@
 import io.vertx.sqlclient.SqlConnection;
 import io.vertx.sqlclient.impl.Connection;
 import io.vertx.sqlclient.impl.PoolBase;
-import io.vertx.sqlclient.impl.SqlConnectionImpl;
 import io.vertx.sqlclient.impl.tracing.QueryTracer;
 import io.vertx.sqlclient.spi.ConnectionFactory;
 
@@ -38,7 +37,7 @@ public MySQLPoolImpl(VertxInternal vertx, MySQLConnectOptions baseConnectOptions
   }
 
   @Override
-  protected SqlConnectionImpl wrap(ContextInternal context, ConnectionFactory factory, Connection conn) {
+  protected MySQLConnectionImpl wrap(ContextInternal context, ConnectionFactory factory, Connection conn) {
     return new MySQLConnectionImpl(context, factory, conn, tracer, metrics);
   }
 
diff --git a/vertx-oracle-client/src/main/java/io/vertx/oracleclient/impl/OracleConnectionFactory.java b/vertx-oracle-client/src/main/java/io/vertx/oracleclient/impl/OracleConnectionFactory.java
@@ -21,7 +21,7 @@
 import io.vertx.sqlclient.PoolOptions;
 import io.vertx.sqlclient.SqlConnection;
 import io.vertx.sqlclient.impl.Connection;
-import io.vertx.sqlclient.impl.SqlConnectionImpl;
+import io.vertx.sqlclient.impl.SqlConnectionBase;
 import io.vertx.sqlclient.impl.tracing.QueryTracer;
 import io.vertx.sqlclient.spi.ConnectionFactory;
 import oracle.jdbc.OracleConnection;
@@ -70,7 +70,7 @@ public Future<SqlConnection> connect(Context context) {
     ContextInternal ic  = (ContextInternal) context;
     return connect(ic)
       .map(c -> {
-        SqlConnectionImpl connection = new SqlConnectionImpl(ic, this, c, OracleDriver.INSTANCE, tracer, metrics);
+        SqlConnectionBase connection = new SqlConnectionBase(ic, this, c, OracleDriver.INSTANCE, tracer, metrics);
         c.init(connection);
         return connection;
       });
diff --git a/vertx-oracle-client/src/main/java/io/vertx/oracleclient/impl/OraclePoolImpl.java b/vertx-oracle-client/src/main/java/io/vertx/oracleclient/impl/OraclePoolImpl.java
@@ -21,13 +21,13 @@
 import io.vertx.sqlclient.Pool;
 import io.vertx.sqlclient.SqlConnection;
 import io.vertx.sqlclient.impl.SqlClientBase;
-import io.vertx.sqlclient.impl.SqlConnectionImpl;
+import io.vertx.sqlclient.impl.SqlConnectionBase;
 import io.vertx.sqlclient.impl.command.CommandBase;
 import io.vertx.sqlclient.impl.tracing.QueryTracer;
 
 import java.util.function.Function;
 
-public class OraclePoolImpl extends SqlClientBase<OraclePoolImpl> implements OraclePool, Closeable {
+public class OraclePoolImpl extends SqlClientBase implements OraclePool, Closeable {
 
   private final OracleConnectionFactory factory;
   private final VertxInternal vertx;
@@ -64,7 +64,7 @@ public Future<SqlConnection> getConnection() {
   private Future<SqlConnection> getConnectionInternal(ContextInternal ctx) {
     return factory.connect(ctx)
       .map(c -> {
-        SqlConnectionImpl<?> connection = new SqlConnectionImpl<>(ctx, factory, c, OracleDriver.INSTANCE, tracer, metrics);
+        SqlConnectionBase<?> connection = new SqlConnectionBase<>(ctx, factory, c, OracleDriver.INSTANCE, tracer, metrics);
         c.init(connection);
         return connection;
       });
@@ -110,6 +110,6 @@ public Future<Void> close() {
   public <R> Future<R> schedule(ContextInternal contextInternal, CommandBase<R> commandBase) {
     ContextInternal ctx = vertx.getOrCreateContext();
     return getConnectionInternal(ctx)
-      .flatMap(conn -> ((SqlConnectionImpl<?>) conn).schedule(ctx, commandBase).eventually(r -> conn.close()));
+      .flatMap(conn -> ((SqlConnectionBase<?>) conn).schedule(ctx, commandBase).eventually(r -> conn.close()));
   }
 }
diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgConnectionImpl.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgConnectionImpl.java
@@ -24,7 +24,7 @@
 import io.vertx.pgclient.spi.PgDriver;
 import io.vertx.sqlclient.impl.Connection;
 import io.vertx.sqlclient.impl.Notification;
-import io.vertx.sqlclient.impl.SqlConnectionImpl;
+import io.vertx.sqlclient.impl.SqlConnectionBase;
 import io.vertx.core.AsyncResult;
 import io.vertx.core.Context;
 import io.vertx.core.Future;
@@ -33,7 +33,7 @@
 import io.vertx.pgclient.impl.codec.TxFailedEvent;
 import io.vertx.sqlclient.impl.tracing.QueryTracer;
 
-public class PgConnectionImpl extends SqlConnectionImpl<PgConnectionImpl> implements PgConnection  {
+public class PgConnectionImpl extends SqlConnectionBase<PgConnectionImpl> implements PgConnection  {
 
   public static Future<PgConnection> connect(ContextInternal context, PgConnectOptions options) {
     if (options.isUsingDomainSocket() && !context.owner().isNativeTransportEnabled()) {
diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgPoolImpl.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgPoolImpl.java
@@ -28,7 +28,6 @@
 import io.vertx.sqlclient.SqlConnection;
 import io.vertx.sqlclient.impl.Connection;
 import io.vertx.sqlclient.impl.PoolBase;
-import io.vertx.sqlclient.impl.SqlConnectionImpl;
 import io.vertx.sqlclient.impl.tracing.QueryTracer;
 import io.vertx.sqlclient.spi.ConnectionFactory;
 
@@ -48,7 +47,7 @@ public PgPoolImpl(VertxInternal vertx, PgConnectOptions baseConnectOptions, Quer
   }
 
   @Override
-  protected SqlConnectionImpl wrap(ContextInternal context, ConnectionFactory factory, Connection conn) {
+  protected PgConnectionImpl wrap(ContextInternal context, ConnectionFactory factory, Connection conn) {
     return new PgConnectionImpl((PgConnectionFactory) factory, context, conn, tracer, metrics);
   }
 
diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/PoolBase.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/PoolBase.java
@@ -43,7 +43,7 @@
  * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
  * @author <a href="mailto:emad.albloushi@gmail.com">Emad Alblueshi</a>
  */
-public abstract class PoolBase<P extends Pool> extends SqlClientBase<P> implements Pool, Closeable {
+public abstract class PoolBase<P extends Pool> extends SqlClientBase implements Pool, Closeable {
 
   private final VertxInternal vertx;
   private final SqlConnectionPool pool;
@@ -147,7 +147,7 @@ public Future<SqlConnection> getConnection() {
       });
     }
     return promise.future().map(conn -> {
-      SqlConnectionImpl wrapper = wrap(current, conn.factory(), conn);
+      SqlConnectionBase wrapper = wrap(current, conn.factory(), conn);
       conn.init(wrapper);
       return wrapper;
     });
@@ -176,7 +176,7 @@ private void acquire(ContextInternal context, long timeout, Handler<AsyncResult<
     pool.acquire(context, timeout, completionHandler);
   }
 
-  protected abstract SqlConnectionImpl wrap(ContextInternal context, ConnectionFactory factory, Connection conn);
+  protected abstract SqlConnectionBase wrap(ContextInternal context, ConnectionFactory factory, Connection conn);
 
   @Override
   public void close(Promise<Void> completion) {
@@ -200,7 +200,7 @@ public Pool connectHandler(Handler<SqlConnection> handler) {
     if (handler != null) {
       connectionInitializer = conn -> {
         ContextInternal current = vertx.getContext();
-        SqlConnectionImpl wrapper = wrap(current, conn.factory(), conn);
+        SqlConnectionBase wrapper = wrap(current, conn.factory(), conn);
         conn.init(wrapper);
         current.dispatch(wrapper, handler);
       };
diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/SqlClientBase.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/SqlClientBase.java
@@ -42,7 +42,7 @@
 import java.util.function.Function;
 import java.util.stream.Collector;
 
-public abstract class SqlClientBase<C extends SqlClient> implements SqlClientInternal, CommandScheduler {
+public abstract class SqlClientBase implements SqlClientInternal, CommandScheduler {
 
   protected final Driver driver;
   protected final QueryTracer tracer;
@@ -191,7 +191,7 @@ public void group(Handler<SqlClient> block) {
     schedule(context(), grouping.composite);
   }
 
-  private class GroupingClient extends SqlClientBase<GroupingClient> {
+  private class GroupingClient extends SqlClientBase {
 
     private CompositeCommand composite = new CompositeCommand();
 
diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/SqlConnectionBase.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/SqlConnectionBase.java
@@ -18,26 +18,34 @@
 package io.vertx.sqlclient.impl;
 
 import io.vertx.core.impl.ContextInternal;
+import io.vertx.core.impl.future.PromiseInternal;
 import io.vertx.core.spi.metrics.ClientMetrics;
 import io.vertx.sqlclient.PrepareOptions;
 import io.vertx.sqlclient.PreparedStatement;
 import io.vertx.sqlclient.SqlClient;
+import io.vertx.sqlclient.SqlConnection;
+import io.vertx.sqlclient.Transaction;
+import io.vertx.sqlclient.impl.command.CommandBase;
 import io.vertx.sqlclient.impl.command.PrepareStatementCommand;
 import io.vertx.core.*;
 import io.vertx.sqlclient.impl.tracing.QueryTracer;
 import io.vertx.sqlclient.spi.ConnectionFactory;
+import io.vertx.sqlclient.spi.DatabaseMetadata;
 import io.vertx.sqlclient.spi.Driver;
 
 /**
  * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
  */
-public abstract class SqlConnectionBase<C extends SqlClient> extends SqlClientBase<C> {
+public class SqlConnectionBase<C extends SqlConnectionBase<C>> extends SqlClientBase implements SqlConnection, Connection.Holder {
 
+  private volatile Handler<Throwable> exceptionHandler;
+  private volatile Handler<Void> closeHandler;
+  protected TransactionImpl tx;
   protected final ContextInternal context;
   protected final ConnectionFactory factory;
   protected final Connection conn;
 
-  protected SqlConnectionBase(ContextInternal context, ConnectionFactory factory, Connection conn, Driver driver, QueryTracer tracer, ClientMetrics metrics) {
+  public SqlConnectionBase(ContextInternal context, ConnectionFactory factory, Connection conn, Driver driver, QueryTracer tracer, ClientMetrics metrics) {
     super(driver, tracer, metrics);
     this.context = context;
     this.factory = factory;
@@ -80,4 +88,117 @@ public C prepare(String sql, Handler<AsyncResult<PreparedStatement>> handler) {
   public Future<PreparedStatement> prepare(String sql) {
     return prepare(sql, (PrepareOptions) null);
   }
+
+  @Override
+  protected ContextInternal context() {
+    return context;
+  }
+
+  @Override
+  protected <T> PromiseInternal<T> promise() {
+    return context.promise();
+  }
+
+  @Override
+  protected <T> PromiseInternal<T> promise(Handler<AsyncResult<T>> handler) {
+    return context.promise(handler);
+  }
+
+  @Override
+  public void handleClosed() {
+    Handler<Void> handler = closeHandler;
+    if (handler != null) {
+      context.emit(handler);
+    }
+  }
+
+  @Override
+  public <R> Future<R> schedule(ContextInternal context, CommandBase<R> cmd) {
+    if (tx != null) {
+      // TODO
+      Promise<R> promise = context.promise();
+      tx.schedule(cmd, promise);
+      return promise.future();
+    } else {
+      return conn.schedule(context, cmd);
+    }
+  }
+
+  @Override
+  public void handleException(Throwable err) {
+    Handler<Throwable> handler = exceptionHandler;
+    if (handler != null) {
+      context.emit(err, handler);
+    } else {
+      err.printStackTrace();
+    }
+  }
+
+  @Override
+  public boolean isSSL() {
+    return conn.isSsl();
+  }
+
+  @Override
+  public DatabaseMetadata databaseMetadata() {
+    return conn.getDatabaseMetaData();
+  }
+
+  @Override
+  public C closeHandler(Handler<Void> handler) {
+    closeHandler = handler;
+    return (C) this;
+  }
+
+  @Override
+  public C exceptionHandler(Handler<Throwable> handler) {
+    exceptionHandler = handler;
+    return (C) this;
+  }
+
+  @Override
+  public Future<Transaction> begin() {
+    if (tx != null) {
+      throw new IllegalStateException();
+    }
+    tx = new TransactionImpl(context, v -> tx = null, conn);
+    return tx.begin();
+  }
+
+  @Override
+  boolean autoCommit() {
+    return tx == null;
+  }
+
+  @Override
+  public void begin(Handler<AsyncResult<Transaction>> handler) {
+    Future<Transaction> fut = begin();
+    fut.onComplete(handler);
+  }
+
+  public void handleEvent(Object event) {
+  }
+
+  @Override
+  public Future<Void> close() {
+    Promise<Void> promise = promise();
+    close(promise);
+    return promise.future();
+  }
+
+  @Override
+  public void close(Handler<AsyncResult<Void>> handler) {
+    close(promise(handler));
+  }
+
+  private void close(Promise<Void> promise) {
+    context.execute(promise, p -> {
+      if (tx != null) {
+        tx.rollback(ar -> conn.close(this, p));
+        tx = null;
+      } else {
+        conn.close(this, p);
+      }
+    });
+  }
 }
diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/SqlConnectionImpl.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/SqlConnectionImpl.java
diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/pool/SqlConnectionPool.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/pool/SqlConnectionPool.java

Original file line number	Diff line number	Diff line change
`@@ -25,7 +25,6 @@`
`25`	`25`	`import io.vertx.sqlclient.SqlConnection;`
`26`	`26`	`import io.vertx.sqlclient.impl.Connection;`
`27`	`27`	`import io.vertx.sqlclient.impl.PoolBase;`
`28`		`-import io.vertx.sqlclient.impl.SqlConnectionImpl;`
`29`	`28`	`import io.vertx.sqlclient.impl.tracing.QueryTracer;`
`30`	`29`	`import io.vertx.sqlclient.spi.ConnectionFactory;`
`31`	`30`
`@@ -38,7 +37,7 @@ public MSSQLPoolImpl(VertxInternal vertx, MSSQLConnectOptions baseConnectOptions`
`38`	`37`	`}`
`39`	`38`
`40`	`39`	`@Override`
`41`		`- protected SqlConnectionImpl wrap(ContextInternal context, ConnectionFactory factory, Connection connection) {`
	`40`	`+ protected MSSQLConnectionImpl wrap(ContextInternal context, ConnectionFactory factory, Connection connection) {`
`42`	`41`	`return new MSSQLConnectionImpl(context, factory, connection, tracer, metrics);`
`43`	`42`	`}`
`44`	`43`
Original file line number	Diff line number	Diff line change
`@@ -28,7 +28,6 @@`
`28`	`28`	`import io.vertx.sqlclient.SqlConnection;`
`29`	`29`	`import io.vertx.sqlclient.impl.Connection;`
`30`	`30`	`import io.vertx.sqlclient.impl.PoolBase;`
`31`		`-import io.vertx.sqlclient.impl.SqlConnectionImpl;`
`32`	`31`	`import io.vertx.sqlclient.impl.tracing.QueryTracer;`
`33`	`32`	`import io.vertx.sqlclient.spi.ConnectionFactory;`
`34`	`33`
`@@ -48,7 +47,7 @@ public PgPoolImpl(VertxInternal vertx, PgConnectOptions baseConnectOptions, Quer`
`48`	`47`	`}`
`49`	`48`
`50`	`49`	`@Override`
`51`		`- protected SqlConnectionImpl wrap(ContextInternal context, ConnectionFactory factory, Connection conn) {`
	`50`	`+ protected PgConnectionImpl wrap(ContextInternal context, ConnectionFactory factory, Connection conn) {`
`52`	`51`	`return new PgConnectionImpl((PgConnectionFactory) factory, context, conn, tracer, metrics);`
`53`	`52`	`}`
`54`	`53`