Skip to content

Commit 0cb53c8

Browse files
SquirySquiry
committed
pgbouncer integration tests
1 parent 64d94a5 commit 0cb53c8

File tree

3 files changed

+142
-17
lines changed

3 files changed

+142
-17
lines changed

‎src/test/java/io/r2dbc/postgresql/client/ReactorNettyClientIntegrationTests.java

Lines changed: 68 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,9 @@
2121
import io.netty.util.ReferenceCountUtil;
2222
import io.r2dbc.postgresql.PostgresqlConnectionConfiguration;
2323
import io.r2dbc.postgresql.PostgresqlConnectionFactory;
24+
import io.r2dbc.postgresql.api.ErrorDetails;
2425
import io.r2dbc.postgresql.api.PostgresqlConnection;
26+
import io.r2dbc.postgresql.api.PostgresqlException;
2527
import io.r2dbc.postgresql.authentication.PasswordAuthenticationHandler;
2628
import io.r2dbc.postgresql.message.backend.BackendMessage;
2729
import io.r2dbc.postgresql.message.backend.CommandComplete;
@@ -30,7 +32,9 @@
3032
import io.r2dbc.postgresql.message.backend.RowDescription;
3133
import io.r2dbc.postgresql.message.frontend.FrontendMessage;
3234
import io.r2dbc.postgresql.message.frontend.Query;
35+
import io.r2dbc.postgresql.util.PgBouncer;
3336
import io.r2dbc.postgresql.util.PostgresqlServerExtension;
37+
import io.r2dbc.spi.R2dbcBadGrammarException;
3438
import io.r2dbc.spi.R2dbcNonTransientResourceException;
3539
import io.r2dbc.spi.R2dbcPermissionDeniedException;
3640
import org.junit.jupiter.api.AfterEach;
@@ -336,29 +340,77 @@ public boolean verify(String s, SSLSession sslSession) {
336340

337341
@Nested
338342
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
339-
final class StatementCacheSizeTests {
343+
final class PgBouncerTests {
340344

341345
@ParameterizedTest
342-
@ValueSource(ints = {0, 2, -1})
343-
void multiplePreparedStatementsTest(int statementCacheSize) {
344-
PostgresqlConnectionFactory connectionFactory = this.createConnectionFactory(statementCacheSize);
346+
@ValueSource(strings = {"transaction", "statement"})
347+
void disabledCacheWorksWithTransactionAndStatementModes(String poolMode) {
348+
try (PgBouncer pgBouncer = new PgBouncer(SERVER, poolMode)) {
349+
PostgresqlConnectionFactory connectionFactory = this.createConnectionFactory(pgBouncer, 0);
350+
351+
connectionFactory.create().flatMapMany(connection -> {
352+
Flux<Integer> q1 = connection.createStatement("SELECT 1 WHERE $1 = 1").bind(0, 1).execute().flatMap(r -> r.map((row, rowMetadata) -> row.get(0, Integer.class)));
353+
Flux<Integer> q2 = connection.createStatement("SELECT 2 WHERE $1 = 2").bind(0, 2).execute().flatMap(r -> r.map((row, rowMetadata) -> row.get(0, Integer.class)));
354+
Flux<Integer> q3 = connection.createStatement("SELECT 3 WHERE $1 = 3").bind(0, 3).execute().flatMap(r -> r.map((row, rowMetadata) -> row.get(0, Integer.class)));
355+
356+
return Flux.concat(q1, q1, q2, q2, q3, q3, connection.close());
357+
})
358+
.as(StepVerifier::create)
359+
.expectNext(1, 1, 2, 2, 3, 3)
360+
.verifyComplete();
361+
}
362+
}
345363

346-
connectionFactory.create().flatMapMany(connection -> {
347-
Flux<Integer> firstQuery = connection.createStatement("SELECT 1 WHERE $1 = 1").bind(0, 1).execute().flatMap(r -> r.map((row, rowMetadata) -> row.get(0, Integer.class)));
348-
Flux<Integer> secondQuery = connection.createStatement("SELECT 2 WHERE $1 = 2").bind(0, 2).execute().flatMap(r -> r.map((row, rowMetadata) -> row.get(0, Integer.class)));
349-
Flux<Integer> thirdQuery = connection.createStatement("SELECT 3 WHERE $1 = 3").bind(0, 3).execute().flatMap(r -> r.map((row, rowMetadata) -> row.get(0, Integer.class)));
364+
@ParameterizedTest
365+
@ValueSource(ints = {-1, 0, 2})
366+
void sessionModeWorksWithAllCaches(int statementCacheSize) {
367+
try (PgBouncer pgBouncer = new PgBouncer(SERVER, "session")) {
368+
PostgresqlConnectionFactory connectionFactory = this.createConnectionFactory(pgBouncer, statementCacheSize);
369+
370+
connectionFactory.create().flatMapMany(connection -> {
371+
Flux<Integer> q1 = connection.createStatement("SELECT 1 WHERE $1 = 1").bind(0, 1).execute().flatMap(r -> r.map((row, rowMetadata) -> row.get(0, Integer.class)));
372+
Flux<Integer> q2 = connection.createStatement("SELECT 2 WHERE $1 = 2").bind(0, 2).execute().flatMap(r -> r.map((row, rowMetadata) -> row.get(0, Integer.class)));
373+
Flux<Integer> q3 = connection.createStatement("SELECT 3 WHERE $1 = 3").bind(0, 3).execute().flatMap(r -> r.map((row, rowMetadata) -> row.get(0, Integer.class)));
374+
375+
return Flux.concat(q1, q1, q2, q2, q3, q3, connection.close());
376+
})
377+
.as(StepVerifier::create)
378+
.expectNext(1, 1, 2, 2, 3, 3)
379+
.verifyComplete();
380+
}
381+
}
350382

351-
return Flux.concat(firstQuery, secondQuery, thirdQuery, connection.close());
352-
})
353-
.as(StepVerifier::create)
354-
.expectNext(1, 2, 3)
355-
.verifyComplete();
383+
@ParameterizedTest
384+
@ValueSource(strings = {"transaction", "statement"})
385+
void statementCacheDoesntWorkWithTransactionAndStatementModes(String poolMode) {
386+
try (PgBouncer pgBouncer = new PgBouncer(SERVER, poolMode)) {
387+
PostgresqlConnectionFactory connectionFactory = this.createConnectionFactory(pgBouncer, -1);
388+
389+
connectionFactory.create().flatMapMany(connection -> {
390+
Flux<Integer> q1 = connection.createStatement("SELECT 1 WHERE $1 = 1").bind(0, 1).execute().flatMap(r -> r.map((row, rowMetadata) -> row.get(0, Integer.class)));
391+
392+
return Flux.concat(q1, q1, connection.close());
393+
})
394+
.as(StepVerifier::create)
395+
.expectNext(1)
396+
.verifyErrorMatches(e -> {
397+
if (!(e instanceof R2dbcBadGrammarException)) {
398+
return false;
399+
}
400+
if (!(e instanceof PostgresqlException)) {
401+
return false;
402+
}
403+
PostgresqlException pgException = (PostgresqlException) e;
404+
ErrorDetails errorDetails = pgException.getErrorDetails();
405+
return errorDetails.getCode().equals("26000") && errorDetails.getMessage().equals("prepared statement \"S_0\" does not exist");
406+
});
407+
}
356408
}
357409

358-
private PostgresqlConnectionFactory createConnectionFactory(int statementCacheSize) {
410+
private PostgresqlConnectionFactory createConnectionFactory(PgBouncer pgBouncer, int statementCacheSize) {
359411
return new PostgresqlConnectionFactory(PostgresqlConnectionConfiguration.builder()
360-
.host(SERVER.getHost())
361-
.port(SERVER.getPort())
412+
.host(pgBouncer.getHost())
413+
.port(pgBouncer.getPort())
362414
.username(SERVER.getUsername())
363415
.password(SERVER.getPassword())
364416
.database(SERVER.getDatabase())

‎src/test/java/io/r2dbc/postgresql/util/PgBouncer.java

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
package io.r2dbc.postgresql.util;
2+
3+
import org.testcontainers.containers.GenericContainer;
4+
import org.testcontainers.containers.PostgreSQLContainer;
5+
import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
6+
7+
public class PgBouncer implements AutoCloseable {
8+
9+
private final GenericContainer<?> container;
10+
11+
public PgBouncer(PostgresqlServerExtension server, String poolMode) {
12+
this.container = new GenericContainer<>("edoburu/pgbouncer")
13+
.withExposedPorts(PostgreSQLContainer.POSTGRESQL_PORT)
14+
.withEnv("POOL_MODE", poolMode)
15+
.withEnv("SERVER_RESET_QUERY_ALWAYS", "1")
16+
.withEnv("DB_USER", server.getUsername())
17+
.withEnv("DB_PASSWORD", server.getPassword())
18+
.withEnv("DB_HOST", server.getPostgres().getNetworkAlias())
19+
.withEnv("DB_PORT", String.valueOf(PostgreSQLContainer.POSTGRESQL_PORT))
20+
.withEnv("DB_NAME", server.getDatabase())
21+
.waitingFor(new HostPortWaitStrategy())
22+
.withNetwork(server.getPostgres().getNetwork());
23+
24+
this.container.start();
25+
}
26+
27+
@Override
28+
public void close() {
29+
this.container.stop();
30+
}
31+
32+
public String getHost() {
33+
return this.container.getContainerIpAddress();
34+
}
35+
36+
public int getPort() {
37+
return this.container.getMappedPort(PostgreSQLContainer.POSTGRESQL_PORT);
38+
}
39+
}

‎src/test/java/io/r2dbc/postgresql/util/PostgresqlServerExtension.java

Lines changed: 35 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.springframework.jdbc.core.JdbcOperations;
2525
import org.springframework.jdbc.core.JdbcTemplate;
2626
import org.testcontainers.containers.JdbcDatabaseContainer;
27+
import org.testcontainers.containers.Network;
2728
import org.testcontainers.containers.PostgreSQLContainer;
2829
import org.testcontainers.utility.MountableFile;
2930
import reactor.util.annotation.Nullable;
@@ -48,13 +49,15 @@
4849
public final class PostgresqlServerExtension implements BeforeAllCallback, AfterAllCallback {
4950

5051
private static PostgreSQLContainer<?> containerInstance = null;
52+
private static Network containerNetwork = null;
5153

5254
private final Supplier<PostgreSQLContainer<?>> container = () -> {
5355

5456
if (PostgresqlServerExtension.containerInstance != null) {
5557
return PostgresqlServerExtension.containerInstance;
5658
}
5759

60+
PostgresqlServerExtension.containerNetwork = Network.newNetwork();
5861
return PostgresqlServerExtension.containerInstance = container();
5962
};
6063

@@ -183,6 +186,9 @@ public String getPassword() {
183186
return this.postgres.getPassword();
184187
}
185188

189+
public DatabaseContainer getPostgres() {
190+
return postgres;
191+
}
186192

187193
private <T extends PostgreSQLContainer<T>> T container() {
188194
T container = new PostgreSQLContainer<T>("postgres:latest")
@@ -193,7 +199,9 @@ private <T extends PostgreSQLContainer<T>> T container() {
193199
.withCopyFileToContainer(getHostPath("setup.sh", 0755), "/var/setup.sh")
194200
.withCopyFileToContainer(getHostPath("test-db-init-script.sql", 0755), "/docker-entrypoint-initdb.d/test-db-init-script.sql")
195201
.withReuse(true)
196-
.withCommand("/var/setup.sh");
202+
.withNetworkAliases("r2dbc-postgres")
203+
.withCommand("/var/setup.sh")
204+
.withNetwork(PostgresqlServerExtension.containerNetwork);
197205

198206
return container;
199207
}
@@ -223,13 +231,18 @@ interface DatabaseContainer {
223231

224232
String getHost();
225233

234+
@Nullable
235+
Network getNetwork();
236+
226237
int getPort();
227238

228239
String getDatabase();
229240

230241
String getUsername();
231242

232243
String getPassword();
244+
245+
String getNetworkAlias();
233246
}
234247

235248
/**
@@ -246,6 +259,12 @@ public String getHost() {
246259
return "localhost";
247260
}
248261

262+
@Override
263+
@Nullable
264+
public Network getNetwork() {
265+
return null;
266+
}
267+
249268
@Override
250269
public int getPort() {
251270
return 5432;
@@ -266,6 +285,11 @@ public String getPassword() {
266285
return "postgres";
267286
}
268287

288+
@Override
289+
public String getNetworkAlias() {
290+
return this.getHost();
291+
}
292+
269293
}
270294

271295
/**
@@ -286,6 +310,11 @@ public String getHost() {
286310
return this.container.getContainerIpAddress();
287311
}
288312

313+
@Override
314+
public Network getNetwork() {
315+
return this.container.getNetwork();
316+
}
317+
289318
@Override
290319
public int getPort() {
291320
return this.container.getMappedPort(PostgreSQLContainer.POSTGRESQL_PORT);
@@ -305,5 +334,10 @@ public String getUsername() {
305334
public String getPassword() {
306335
return this.container.getPassword();
307336
}
337+
338+
@Override
339+
public String getNetworkAlias() {
340+
return "r2dbc-postgres";
341+
}
308342
}
309343
}

0 commit comments

Comments
 (0)