Skip to content

Commit 78db24d

Browse files
committed
Adapt to pool changes in vertx-core.
1 parent 81cbcd3 commit 78db24d

File tree

2 files changed

+23
-29
lines changed

2 files changed

+23
-29
lines changed

vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/PreparedStatementImpl.java

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -246,13 +246,9 @@ private void execute(Tuple args, PromiseInternal<R> promise) {
246246
PreparedStatementImpl.this.execute(args, 0, null, false, builder, promise);
247247
}
248248

249-
public void executeBatch(List<Tuple> argsList, Handler<AsyncResult<R>> handler) {
250-
executeBatch(argsList, context.promise(handler));
251-
}
252-
253249
@Override
254250
public Future<R> executeBatch(List<Tuple> argsList) {
255-
Promise<R> promise = context.promise();
251+
PromiseInternal<R> promise = context.promise();
256252
executeBatch(argsList, promise);
257253
return promise.future();
258254
}

vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/pool/SqlConnectionPool.java

Lines changed: 22 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -103,27 +103,25 @@ public EventLoopContext apply(ContextInternal contextInternal) {
103103

104104
private final PoolConnector<PooledConnection> connector = new PoolConnector<PooledConnection>() {
105105
@Override
106-
public void connect(EventLoopContext context, PoolConnector.Listener listener, Handler<AsyncResult<ConnectResult<PooledConnection>>> handler) {
106+
public Future<ConnectResult<PooledConnection>> connect(EventLoopContext context, Listener listener) {
107107
Future<SqlConnection> future = connectionProvider.apply(context);
108-
future.onComplete(ar -> {
109-
if (ar.succeeded()) {
110-
SqlConnectionBase res = (SqlConnectionBase) ar.result();
111-
Connection conn = res.unwrap();
112-
if (conn.isValid()) {
113-
PooledConnection pooled = new PooledConnection(res.factory(), conn, listener);
114-
conn.init(pooled);
115-
Handler<PooledConnection> connectionHandler = hook.get();
116-
if (connectionHandler != null) {
117-
pooled.poolResultHandler = handler;
118-
connectionHandler.handle(pooled);
119-
} else {
120-
handler.handle(Future.succeededFuture(new ConnectResult<>(pooled, pipeliningLimit, 0)));
121-
}
108+
return future.compose(res -> {
109+
SqlConnectionBase connBase = (SqlConnectionBase) res;
110+
Connection conn = connBase.unwrap();
111+
if (conn.isValid()) {
112+
PooledConnection pooled = new PooledConnection(connBase.factory(), conn, listener);
113+
conn.init(pooled);
114+
Handler<PooledConnection> connectionHandler = hook.get();
115+
if (connectionHandler != null) {
116+
Promise<ConnectResult<PooledConnection>> p = Promise.promise();
117+
pooled.poolCallback = p;
118+
connectionHandler.handle(pooled);
119+
return p.future();
122120
} else {
123-
handler.handle(Future.failedFuture(ConnectionBase.CLOSED_EXCEPTION));
121+
return Future.succeededFuture(new ConnectResult<>(pooled, pipeliningLimit, 0));
124122
}
125123
} else {
126-
handler.handle(Future.failedFuture(ar.cause()));
124+
return Future.failedFuture(ConnectionBase.CLOSED_EXCEPTION);
127125
}
128126
});
129127
}
@@ -261,7 +259,7 @@ public class PooledConnection implements Connection, Connection.Holder {
261259
private final Connection conn;
262260
private final PoolConnector.Listener listener;
263261
private Holder holder;
264-
private Handler<AsyncResult<ConnectResult<PooledConnection>>> poolResultHandler;
262+
private Promise<ConnectResult<PooledConnection>> poolCallback;
265263
private Lease<PooledConnection> lease;
266264
public long expirationTimestamp;
267265

@@ -332,11 +330,11 @@ private void doClose(Holder holder, Promise<Void> promise) {
332330
promise.fail(msg);
333331
} else {
334332
this.holder = null;
335-
Handler<AsyncResult<ConnectResult<PooledConnection>>> resultHandler = poolResultHandler;
333+
Promise<ConnectResult<PooledConnection>> resultHandler = poolCallback;
336334
if (resultHandler != null) {
337-
poolResultHandler = null;
335+
poolCallback = null;
338336
promise.complete();
339-
resultHandler.handle(Future.succeededFuture(new ConnectResult<>(this, pipeliningLimit, 0)));
337+
resultHandler.complete(new ConnectResult<>(this, pipeliningLimit, 0));
340338
return;
341339
}
342340
if (beforeRecycle == null) {
@@ -360,10 +358,10 @@ public void handleClosed() {
360358
if (holder != null) {
361359
holder.handleClosed();
362360
}
363-
Handler<AsyncResult<ConnectResult<PooledConnection>>> resultHandler = poolResultHandler;
361+
Promise<ConnectResult<PooledConnection>> resultHandler = poolCallback;
364362
if (resultHandler != null) {
365-
poolResultHandler = null;
366-
resultHandler.handle(Future.failedFuture(ConnectionBase.CLOSED_EXCEPTION));
363+
poolCallback = null;
364+
resultHandler.fail(ConnectionBase.CLOSED_EXCEPTION);
367365
}
368366
listener.onRemove();
369367
}

0 commit comments

Comments
 (0)