diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgEncoder.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgEncoder.java index c47b1723a..8d58b2270 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgEncoder.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgEncoder.java @@ -23,13 +23,15 @@ import io.netty.channel.ChannelOutboundHandlerAdapter; import io.netty.channel.ChannelPromise; import io.netty.channel.socket.SocketChannel; -import io.vertx.sqlclient.Tuple; +import io.netty.util.concurrent.PromiseNotifier; import io.vertx.pgclient.impl.util.Util; +import io.vertx.sqlclient.Tuple; import io.vertx.sqlclient.impl.HexSequence; import io.vertx.sqlclient.internal.RowDescriptorBase; import io.vertx.sqlclient.spi.protocol.CloseConnectionCommand; -import java.util.*; +import java.util.ArrayList; +import java.util.Map; import static io.vertx.pgclient.impl.util.Util.writeCString; import static java.nio.charset.StandardCharsets.UTF_8; @@ -98,6 +100,9 @@ public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exce CloseConnectionCommand cmd = CloseConnectionCommand.INSTANCE; PgCommandMessage codec = PgCommandMessage.wrap(cmd); codec.encode(this); + ctx.close(promise); + } else { + ctx.channel().closeFuture().addListener(new PromiseNotifier<>(promise)); } } diff --git a/vertx-pg-client/src/test/java/io/vertx/tests/pgclient/CloseConnectionTest.java b/vertx-pg-client/src/test/java/io/vertx/tests/pgclient/CloseConnectionTest.java index 930db8f0c..cbb3287c9 100644 --- a/vertx-pg-client/src/test/java/io/vertx/tests/pgclient/CloseConnectionTest.java +++ b/vertx-pg-client/src/test/java/io/vertx/tests/pgclient/CloseConnectionTest.java @@ -1,12 +1,16 @@ package io.vertx.tests.pgclient; import io.netty.buffer.ByteBufUtil; +import io.netty.channel.ChannelOption; import io.vertx.core.Vertx; import io.vertx.core.buffer.Buffer; +import io.vertx.core.internal.net.NetSocketInternal; import io.vertx.ext.unit.Async; import io.vertx.ext.unit.TestContext; import io.vertx.pgclient.PgBuilder; +import io.vertx.pgclient.PgConnectOptions; import io.vertx.pgclient.PgConnection; +import io.vertx.pgclient.SslMode; import io.vertx.sqlclient.Pool; import io.vertx.sqlclient.PoolOptions; import io.vertx.tests.sqlclient.ProxyServer; @@ -14,6 +18,9 @@ import org.junit.Before; import org.junit.Test; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.TimeUnit; + public class CloseConnectionTest extends PgTestBase { protected Vertx vertx; @@ -74,6 +81,51 @@ private void testCloseConnection(TestContext ctx, Runnable test) { })); } + @Test + public void testReadIdleTimeoutClosesConnection(TestContext ctx) { + Async async = ctx.async(); + vertx.createNetServer().connectHandler(socket -> { + // Enable half-close so the server keeps the connection open after receiving FIN, + // otherwise Vert.x auto-closes and sends FIN back, masking the bug + ((NetSocketInternal) socket).channelHandlerContext().channel().config() + .setOption(ChannelOption.ALLOW_HALF_CLOSURE, true); + socket.handler(data -> { + // Received startup message, send PG handshake response then stay silent + Buffer response = Buffer.buffer() + .appendByte((byte) 'R').appendInt(8).appendInt(0) + .appendByte((byte) 'S').appendInt(25) + .appendBytes("client_encoding\0UTF8\0".getBytes(StandardCharsets.UTF_8)) + .appendByte((byte) 'K').appendInt(12).appendInt(1).appendInt(0) + .appendByte((byte) 'Z').appendInt(5).appendByte((byte) 'I'); + socket.write(response); + socket.handler(null); + }); + }).listen(0, "localhost").onComplete(ctx.asyncAssertSuccess(server -> { + PgConnectOptions connectOptions = new PgConnectOptions() + .setHost("localhost") + .setPort(server.actualPort()) + .setUser("test") + .setPassword("test") + .setDatabase("test") + .setSslMode(SslMode.DISABLE); + Pool pool = PgBuilder.pool() + .connectingTo(connectOptions) + .with(new PoolOptions().setMaxSize(1)) + .with(new io.vertx.core.net.NetClientOptions() + .setReadIdleTimeout(200) + .setIdleTimeoutUnit(TimeUnit.MILLISECONDS)) + .using(vertx) + .build(); + pool.getConnection().onComplete(ctx.asyncAssertSuccess(conn -> + conn.closeHandler(v -> { + pool.close(); + server.close(); + async.complete(); + }) + )); + })); + } + @Test public void testTransactionInProgressShouldFail(TestContext ctx) { ProxyServer proxy = ProxyServer.create(vertx, options.getPort(), options.getHost()); diff --git a/vertx-pg-client/src/test/java/io/vertx/tests/pgclient/tck/PgPreparedQueryTestBase.java b/vertx-pg-client/src/test/java/io/vertx/tests/pgclient/tck/PgPreparedQueryTestBase.java index 69f243cc0..a79d43470 100644 --- a/vertx-pg-client/src/test/java/io/vertx/tests/pgclient/tck/PgPreparedQueryTestBase.java +++ b/vertx-pg-client/src/test/java/io/vertx/tests/pgclient/tck/PgPreparedQueryTestBase.java @@ -52,10 +52,9 @@ public void testMaximumParametersExceeded(TestContext ctx) { connect(ctx.asyncAssertSuccess(conn -> { conn.preparedQuery(query) .execute(tuple) - .onComplete(ar -> { - ctx.assertTrue(ar.failed()); - ctx.assertTrue(ar.cause().getMessage().contains("exceeds the maximum of 65535")); - }); + .onComplete(ctx.asyncAssertFailure(t -> { + ctx.assertTrue(t.getMessage().contains("exceeds the maximum of 65535")); + })); })); } }