Skip to content

Commit 91e69d1

Browse files
committed
Implement Postgres request cancellation
1 parent f854f10 commit 91e69d1

File tree

5 files changed

+9
-8
lines changed

5 files changed

+9
-8
lines changed

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

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -97,9 +97,8 @@ protected Future<Connection> doConnectInternal(SqlConnectOptions options, EventL
9797
});
9898
}
9999

100-
public void cancelRequest(SocketAddress server, int processId, int secretKey, Handler<AsyncResult<Void>> handler) {
101-
// NOT GOOD
102-
doConnect(server, vertx.createEventLoopContext(), (PgConnectOptions) options).onComplete(ar -> {
100+
public void cancelRequest(PgConnectOptions options, int processId, int secretKey, Handler<AsyncResult<Void>> handler) {
101+
doConnect(options.getSocketAddress(), vertx.createEventLoopContext(), options).onComplete(ar -> {
103102
if (ar.succeeded()) {
104103
PgSocketConnection conn = (PgSocketConnection) ar.result();
105104
conn.sendCancelRequestMessage(processId, secretKey, handler);
@@ -182,6 +181,8 @@ private PgSocketConnection newSocketConnection(EventLoopContext context, NetSock
182181
Predicate<String> preparedStatementCacheSqlFilter = options.getPreparedStatementCacheSqlFilter();
183182
int pipeliningLimit = options.getPipeliningLimit();
184183
boolean useLayer7Proxy = options.getUseLayer7Proxy();
185-
return new PgSocketConnection(socket, cachePreparedStatements, preparedStatementCacheMaxSize, preparedStatementCacheSqlFilter, pipeliningLimit, useLayer7Proxy, context);
184+
PgSocketConnection conn = new PgSocketConnection(socket, cachePreparedStatements, preparedStatementCacheMaxSize, preparedStatementCacheSqlFilter, pipeliningLimit, useLayer7Proxy, context);
185+
conn.options = options;
186+
return conn;
186187
}
187188
}

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -133,7 +133,8 @@ public Future<Void> cancelRequest() {
133133
public PgConnection cancelRequest(Handler<AsyncResult<Void>> handler) {
134134
Context current = Vertx.currentContext();
135135
if (current == context) {
136-
((PgConnectionFactory)factory).cancelRequest(conn.server(), this.processId(), this.secretKey(), handler);
136+
PgSocketConnection unwrap = (PgSocketConnection) conn.unwrap();
137+
((PgConnectionFactory)factory).cancelRequest(unwrap.options, this.processId(), this.secretKey(), handler);
137138
} else {
138139
context.runOnContext(v -> cancelRequest(handler));
139140
}

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import io.vertx.core.buffer.Buffer;
2727
import io.vertx.core.impl.EventLoopContext;
2828
import io.vertx.core.net.impl.NetSocketInternal;
29+
import io.vertx.pgclient.PgConnectOptions;
2930
import io.vertx.pgclient.PgException;
3031
import io.vertx.pgclient.impl.codec.NoticeResponse;
3132
import io.vertx.pgclient.impl.codec.PgCodec;
@@ -54,6 +55,7 @@ public class PgSocketConnection extends SocketConnectionBase {
5455
public int processId;
5556
public int secretKey;
5657
public PgDatabaseMetadata dbMetaData;
58+
PgConnectOptions options;
5759

5860
public PgSocketConnection(NetSocketInternal socket,
5961
boolean cachePreparedStatements,

vertx-pg-client/src/test/java/io/vertx/pgclient/PgConnectionTest.java

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -107,7 +107,6 @@ public void testQueueQueries(TestContext ctx) {
107107
}));
108108
}
109109

110-
@Ignore("FIXME")
111110
@Test
112111
public void testCancelRequest(TestContext ctx) {
113112
Async async = ctx.async(2);

vertx-pg-client/src/test/java/io/vertx/pgclient/PgPoolTestBase.java

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,6 @@
2424
import io.vertx.sqlclient.*;
2525
import org.junit.After;
2626
import org.junit.Before;
27-
import org.junit.Ignore;
2827
import org.junit.Test;
2928

3029
import java.util.concurrent.atomic.AtomicReference;
@@ -210,7 +209,6 @@ public void testReconnect(TestContext ctx) {
210209
}));
211210
}
212211

213-
@Ignore("FIXME")
214212
@Test
215213
public void testCancelRequest(TestContext ctx) {
216214
Async async = ctx.async();

0 commit comments

Comments
 (0)