Skip to content

Commit 2fc3af2

Browse files
mp911demp911de
committed
Refactor DisposableServer into JUnit extension.
Use parameter injection to avoid resource control in tests. [#678][#677]
1 parent 8d1f297 commit 2fc3af2

File tree

4 files changed

+205
-81
lines changed

4 files changed

+205
-81
lines changed

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

Lines changed: 11 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -19,12 +19,11 @@
1919
import io.r2dbc.postgresql.PostgresqlConnectionConfiguration;
2020
import io.r2dbc.postgresql.PostgresqlConnectionFactory;
2121
import io.r2dbc.postgresql.api.PostgresqlException;
22+
import io.r2dbc.postgresql.util.Disposable;
2223
import org.junit.jupiter.api.Test;
23-
import reactor.netty.DisposableChannel;
24-
import reactor.netty.DisposableServer;
25-
import reactor.netty.tcp.TcpServer;
2624
import reactor.test.StepVerifier;
2725

26+
import java.net.InetSocketAddress;
2827
import java.nio.channels.ClosedChannelException;
2928
import java.util.function.Consumer;
3029

@@ -33,16 +32,16 @@
3332
public class DowntimeIntegrationTests {
3433

3534
@Test
36-
void failSslHandshakeIfInboundClosed() {
37-
verifyError(SSLMode.REQUIRE, error ->
35+
void failSslHandshakeIfInboundClosed(@Disposable InetSocketAddress faultyServer) {
36+
verifyError(faultyServer, SSLMode.REQUIRE, error ->
3837
assertThat(error)
3938
.isInstanceOf(AbstractPostgresSSLHandlerAdapter.PostgresqlSslException.class)
4039
.hasMessage("Connection closed during SSL negotiation"));
4140
}
4241

4342
@Test
44-
void failSslTunnelIfInboundClosed() {
45-
verifyError(SSLMode.TUNNEL, error -> {
43+
void failSslTunnelIfInboundClosed(@Disposable InetSocketAddress faultyServer) {
44+
verifyError(faultyServer, SSLMode.TUNNEL, error -> {
4645
assertThat(error)
4746
.isInstanceOf(PostgresqlException.class)
4847
.cause()
@@ -55,31 +54,19 @@ void failSslTunnelIfInboundClosed() {
5554
});
5655
}
5756

58-
// Simulate server downtime, where connections are accepted and then closed immediately
59-
static DisposableServer newServer() {
60-
return TcpServer.create()
61-
.doOnConnection(DisposableChannel::dispose)
62-
.bindNow();
63-
}
64-
65-
static PostgresqlConnectionFactory newConnectionFactory(DisposableServer server, SSLMode sslMode) {
57+
static PostgresqlConnectionFactory newConnectionFactory(InetSocketAddress server, SSLMode sslMode) {
6658
return new PostgresqlConnectionFactory(
6759
PostgresqlConnectionConfiguration.builder()
68-
.host(server.host())
69-
.port(server.port())
60+
.host(server.getHostString())
61+
.port(server.getPort())
7062
.username("test")
7163
.sslMode(sslMode)
7264
.build());
7365
}
7466

75-
static void verifyError(SSLMode sslMode, Consumer<Throwable> assertions) {
76-
DisposableServer server = newServer();
67+
static void verifyError(InetSocketAddress server, SSLMode sslMode, Consumer<Throwable> assertions) {
7768
PostgresqlConnectionFactory connectionFactory = newConnectionFactory(server, sslMode);
78-
try {
79-
connectionFactory.create().as(StepVerifier::create).verifyErrorSatisfies(assertions);
80-
} finally {
81-
server.disposeNow();
82-
}
69+
connectionFactory.create().as(StepVerifier::create).verifyErrorSatisfies(assertions);
8370
}
8471

8572
}

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

Lines changed: 62 additions & 57 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
import io.r2dbc.postgresql.PostgresqlConnectionConfiguration;
2121
import io.r2dbc.postgresql.PostgresqlConnectionFactory;
2222
import io.r2dbc.postgresql.api.PostgresqlConnection;
23+
import io.r2dbc.postgresql.util.Disposable;
2324
import io.r2dbc.postgresql.util.PostgresqlHighAvailabilityClusterExtension;
2425
import io.r2dbc.spi.Connection;
2526
import io.r2dbc.spi.R2dbcException;
@@ -29,11 +30,11 @@
2930
import org.testcontainers.containers.PostgreSQLContainer;
3031
import reactor.core.publisher.Flux;
3132
import reactor.core.publisher.Mono;
32-
import reactor.netty.DisposableChannel;
33-
import reactor.netty.DisposableServer;
34-
import reactor.netty.tcp.TcpServer;
3533
import reactor.test.StepVerifier;
3634

35+
import java.net.InetSocketAddress;
36+
import java.util.function.Consumer;
37+
3738
import static org.assertj.core.api.Assertions.assertThat;
3839

3940
/**
@@ -52,7 +53,7 @@ void testPrimaryAndStandbyStartup() {
5253

5354
@Test
5455
void testMultipleCallsOnSameFactory() {
55-
PostgresqlConnectionFactory connectionFactory = this.multiHostConnectionFactory(MultiHostConnectionStrategy.TargetServerType.PREFER_SECONDARY, SERVERS.getPrimary(), SERVERS.getStandby());
56+
PostgresqlConnectionFactory connectionFactory = this.configure(MultiHostConnectionStrategy.TargetServerType.PREFER_SECONDARY, SERVERS.getPrimary(), SERVERS.getStandby());
5657

5758
Mono.usingWhen(connectionFactory.create(), this::isPrimary, Connection::close)
5859
.as(StepVerifier::create)
@@ -124,21 +125,16 @@ void testTargetPreferSecondaryConnectedToStandby() {
124125
}
125126

126127
@Test
127-
void testTargetPreferSecondaryConnectedToMasterOnStandbyFailure() {
128-
DisposableServer failingServer = newServer();
129-
try {
130-
isConnectedToPrimary(MultiHostConnectionStrategy.TargetServerType.PREFER_SECONDARY, SERVERS.getPrimary(), failingServer)
131-
.as(StepVerifier::create)
132-
.expectNext(true)
133-
.verifyComplete();
134-
} finally {
135-
failingServer.dispose();
136-
}
128+
void testTargetPreferSecondaryConnectedToMasterOnStandbyFailure(@Disposable InetSocketAddress faulty) {
129+
isConnectedToPrimary(MultiHostConnectionStrategy.TargetServerType.PREFER_SECONDARY, SERVERS.getPrimary(), faulty)
130+
.as(StepVerifier::create)
131+
.expectNext(true)
132+
.verifyComplete();
137133
}
138134

139135
@Test
140136
void testMultipleCallsWithTargetPreferSecondaryConnectedToStandby() {
141-
PostgresqlConnectionFactory connectionFactory = this.multiHostConnectionFactory(MultiHostConnectionStrategy.TargetServerType.PREFER_SECONDARY, SERVERS.getPrimary(), SERVERS.getStandby());
137+
PostgresqlConnectionFactory connectionFactory = this.configure(MultiHostConnectionStrategy.TargetServerType.PREFER_SECONDARY, SERVERS.getPrimary(), SERVERS.getStandby());
142138

143139
Mono<Boolean> allocator = Mono.usingWhen(connectionFactory.create(), this::isPrimary, Connection::close);
144140
Flux<Boolean> connectionPool = Flux.merge(allocator, allocator);
@@ -151,23 +147,28 @@ void testMultipleCallsWithTargetPreferSecondaryConnectedToStandby() {
151147
}
152148

153149
@Test
154-
void testMultipleCallsWithTargetPreferSecondaryConnectedToMasterOnStandbyFailure() {
155-
DisposableServer failingServer = newServer();
156-
try {
157-
PostgresqlConnectionFactory connectionFactory = this.multiHostConnectionFactoryWithFailingServer(MultiHostConnectionStrategy.TargetServerType.PREFER_SECONDARY, SERVERS.getPrimary(),
158-
failingServer);
159-
160-
Mono<Boolean> allocator = Mono.usingWhen(connectionFactory.create(), this::isPrimary, Connection::close);
161-
Flux<Boolean> connectionPool = Flux.merge(allocator, allocator);
162-
163-
connectionPool
164-
.as(StepVerifier::create)
165-
.expectNext(true)
166-
.expectNext(true)
167-
.verifyComplete();
168-
} finally {
169-
failingServer.dispose();
170-
}
150+
void testAllFaulty(@Disposable InetSocketAddress faulty1, @Disposable InetSocketAddress faulty2) {
151+
PostgresqlConnectionFactory connectionFactory = this.configure(MultiHostConnectionStrategy.TargetServerType.SECONDARY, SERVERS.getPrimary(),
152+
faulty1, faulty2);
153+
154+
connectionFactory.create()
155+
.as(StepVerifier::create)
156+
.expectError(R2dbcNonTransientResourceException.class);
157+
}
158+
159+
@Test
160+
void testMultipleCallsWithTargetPreferSecondaryConnectedToMasterOnStandbyFailure(@Disposable InetSocketAddress faulty) {
161+
PostgresqlConnectionFactory connectionFactory = this.configure(MultiHostConnectionStrategy.TargetServerType.PREFER_SECONDARY, SERVERS.getPrimary(),
162+
faulty);
163+
164+
Mono<Boolean> allocator = Mono.usingWhen(connectionFactory.create(), this::isPrimary, Connection::close);
165+
Flux<Boolean> connectionPool = Flux.merge(allocator, allocator);
166+
167+
connectionPool
168+
.as(StepVerifier::create)
169+
.expectNext(true)
170+
.expectNext(true)
171+
.verifyComplete();
171172
}
172173

173174
@Test
@@ -227,13 +228,13 @@ void testTargetSecondaryFailedOnPrimary() {
227228
}
228229

229230
private Mono<Boolean> isConnectedToPrimary(MultiHostConnectionStrategy.TargetServerType targetServerType, PostgreSQLContainer<?>... servers) {
230-
PostgresqlConnectionFactory connectionFactory = this.multiHostConnectionFactory(targetServerType, servers);
231+
PostgresqlConnectionFactory connectionFactory = this.configure(targetServerType, servers);
231232

232233
return Mono.usingWhen(connectionFactory.create(), this::isPrimary, Connection::close);
233234
}
234235

235-
private Mono<Boolean> isConnectedToPrimary(MultiHostConnectionStrategy.TargetServerType targetServerType, PostgreSQLContainer<?> primaryServer, DisposableServer failingServer) {
236-
PostgresqlConnectionFactory connectionFactory = this.multiHostConnectionFactoryWithFailingServer(targetServerType, primaryServer, failingServer);
236+
private Mono<Boolean> isConnectedToPrimary(MultiHostConnectionStrategy.TargetServerType targetServerType, PostgreSQLContainer<?> primaryServer, InetSocketAddress failingServer) {
237+
PostgresqlConnectionFactory connectionFactory = this.configure(targetServerType, primaryServer, failingServer);
237238

238239
return Mono.usingWhen(connectionFactory.create(), this::isPrimary, Connection::close);
239240
}
@@ -246,25 +247,36 @@ private Mono<Boolean> isPrimary(PostgresqlConnection connection) {
246247
.next();
247248
}
248249

249-
private PostgresqlConnectionFactory multiHostConnectionFactory(MultiHostConnectionStrategy.TargetServerType targetServerType, PostgreSQLContainer<?>... servers) {
250-
PostgreSQLContainer<?> firstServer = servers[0];
251-
PostgresqlConnectionConfiguration.Builder builder = PostgresqlConnectionConfiguration.builder();
252-
for (PostgreSQLContainer<?> server : servers) {
253-
builder.addHost(server.getHost(), server.getMappedPort(5432));
254-
}
255-
PostgresqlConnectionConfiguration configuration = builder
256-
.targetServerType(targetServerType)
257-
.username(firstServer.getUsername())
258-
.password(firstServer.getPassword())
259-
.build();
260-
return new PostgresqlConnectionFactory(configuration);
250+
private PostgresqlConnectionFactory configure(MultiHostConnectionStrategy.TargetServerType targetServerType, PostgreSQLContainer<?>... servers) {
251+
return configure(targetServerType, servers[0], builder -> {
252+
253+
254+
for (PostgreSQLContainer<?> server : servers) {
255+
256+
if (server == servers[0]) {
257+
continue;
258+
}
259+
builder.addHost(server.getHost(), server.getMappedPort(5432));
260+
}
261+
});
262+
}
263+
264+
private PostgresqlConnectionFactory configure(MultiHostConnectionStrategy.TargetServerType targetServerType, PostgreSQLContainer<?> primaryServer,
265+
InetSocketAddress... addresses) {
266+
267+
return configure(targetServerType, primaryServer, builder -> {
268+
269+
for (InetSocketAddress address : addresses) {
270+
builder.addHost(address.getHostName(), address.getPort());
271+
}
272+
});
261273
}
262274

263-
private PostgresqlConnectionFactory multiHostConnectionFactoryWithFailingServer(MultiHostConnectionStrategy.TargetServerType targetServerType, PostgreSQLContainer<?> primaryServer,
264-
DisposableServer failingServer) {
275+
private PostgresqlConnectionFactory configure(MultiHostConnectionStrategy.TargetServerType targetServerType, PostgreSQLContainer<?> primaryServer,
276+
Consumer<PostgresqlConnectionConfiguration.Builder> builderCustomizer) {
265277
PostgresqlConnectionConfiguration.Builder builder = PostgresqlConnectionConfiguration.builder();
266278
builder.addHost(primaryServer.getHost(), primaryServer.getMappedPort(5432));
267-
builder.addHost(failingServer.host(), failingServer.port());
279+
builderCustomizer.accept(builder);
268280

269281
PostgresqlConnectionConfiguration configuration = builder
270282
.targetServerType(targetServerType)
@@ -274,11 +286,4 @@ private PostgresqlConnectionFactory multiHostConnectionFactoryWithFailingServer(
274286
return new PostgresqlConnectionFactory(configuration);
275287
}
276288

277-
// Simulate server downtime, where connections are accepted and then closed immediately
278-
static DisposableServer newServer() {
279-
return TcpServer.create()
280-
.doOnConnection(DisposableChannel::dispose)
281-
.bindNow();
282-
}
283-
284289
}

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

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
1+
/*
2+
* Copyright 2025 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package io.r2dbc.postgresql.util;
18+
19+
import org.junit.jupiter.api.extension.ExtendWith;
20+
21+
import java.lang.annotation.Documented;
22+
import java.lang.annotation.ElementType;
23+
import java.lang.annotation.Retention;
24+
import java.lang.annotation.RetentionPolicy;
25+
import java.lang.annotation.Target;
26+
import java.net.InetSocketAddress;
27+
28+
/**
29+
* Extension for a disposable server that does not reply, only accepting connections and closing these immediately.
30+
* The extension injects a {@link InetSocketAddress} that points to the disposable server.
31+
* <p>
32+
* Example usage:
33+
*
34+
* <pre class="code">
35+
* public class ExampleTest {
36+
*
37+
* &#064;Test
38+
* public void shouldDoSomething(@Disposable InetSocketAddress address) {
39+
* // ... connect to address
40+
* }
41+
* }
42+
* </pre>
43+
*
44+
* @author Mark Paluch
45+
*/
46+
@Target({ElementType.PARAMETER})
47+
@Retention(RetentionPolicy.RUNTIME)
48+
@Documented
49+
@ExtendWith({DisposableServerExtension.class})
50+
public @interface Disposable {
51+
52+
}

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

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,80 @@
1+
/*
2+
* Copyright 2017 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package io.r2dbc.postgresql.util;
18+
19+
import org.junit.jupiter.api.extension.AfterAllCallback;
20+
import org.junit.jupiter.api.extension.ExtensionContext;
21+
import org.junit.jupiter.api.extension.ParameterContext;
22+
import org.junit.jupiter.api.extension.ParameterResolutionException;
23+
import org.junit.jupiter.api.extension.ParameterResolver;
24+
import reactor.netty.DisposableChannel;
25+
import reactor.netty.DisposableServer;
26+
import reactor.netty.tcp.TcpServer;
27+
28+
import java.lang.reflect.Parameter;
29+
import java.net.InetSocketAddress;
30+
import java.util.Map;
31+
import java.util.concurrent.ConcurrentHashMap;
32+
33+
/**
34+
* JUnit Extension to create a disposable server that does not reply, only accepting connections.
35+
*/
36+
final class DisposableServerExtension implements ParameterResolver, AfterAllCallback {
37+
38+
private static final ExtensionContext.Namespace NAMESPACE = ExtensionContext.Namespace.create(DisposableServerExtension.class);
39+
40+
@Override
41+
public boolean supportsParameter(ParameterContext parameterContext, ExtensionContext extensionContext) throws ParameterResolutionException {
42+
return parameterContext.isAnnotated(Disposable.class) && parameterContext.getParameter().getType().isAssignableFrom(InetSocketAddress.class);
43+
}
44+
45+
@Override
46+
@SuppressWarnings("unchecked")
47+
public Object resolveParameter(ParameterContext parameterContext, ExtensionContext extensionContext) throws ParameterResolutionException {
48+
49+
ExtensionContext.Store store = extensionContext.getStore(NAMESPACE);
50+
Map<Parameter, DisposableServer> servers = store.getOrComputeIfAbsent("servers", key -> new ConcurrentHashMap<>(), Map.class);
51+
52+
53+
DisposableServer server = servers.computeIfAbsent(parameterContext.getParameter(), key -> newServer());
54+
return InetSocketAddress.createUnresolved(server.host(), server.port());
55+
}
56+
57+
@Override
58+
@SuppressWarnings("unchecked")
59+
public void afterAll(ExtensionContext extensionContext) {
60+
61+
ExtensionContext.Store store = extensionContext.getStore(NAMESPACE);
62+
Map<Parameter, DisposableServer> servers = store.get("servers", Map.class);
63+
64+
if (servers != null) {
65+
for (DisposableServer server : servers.values()) {
66+
server.disposeNow();
67+
}
68+
servers.clear();
69+
}
70+
71+
}
72+
73+
static DisposableServer newServer() {
74+
return TcpServer.create()
75+
.doOnConnection(DisposableChannel::dispose)
76+
.bindNow();
77+
}
78+
79+
80+
}

0 commit comments

Comments
 (0)