PostgresChannelMessageTableSubscriber: Renew connection only if invalid
Fixes: #9111
An evolution of the #9061: renew the connection only when we need to.
(cherry picked from commit da29e2da6a)
This commit is contained in:
committed by
Spring Builds
parent
5a5f2dec82
commit
ace6cfe9b0
@@ -23,8 +23,10 @@ import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
import java.util.function.Consumer;
|
||||
|
||||
import javax.sql.DataSource;
|
||||
|
||||
@@ -270,7 +272,18 @@ public class PostgresChannelMessageTableSubscriberTests implements PostgresConta
|
||||
CountDownLatch latch = new CountDownLatch(2);
|
||||
List<Object> payloads = new ArrayList<>();
|
||||
CountDownLatch connectionLatch = new CountDownLatch(2);
|
||||
connectionSupplier.onGetConnection = connectionLatch::countDown;
|
||||
AtomicBoolean connectionCloseState = new AtomicBoolean();
|
||||
connectionSupplier.onGetConnection = conn -> {
|
||||
connectionLatch.countDown();
|
||||
if (connectionCloseState.compareAndSet(false, true)) {
|
||||
try {
|
||||
conn.close();
|
||||
}
|
||||
catch (Exception e) {
|
||||
//nop
|
||||
}
|
||||
}
|
||||
};
|
||||
postgresChannelMessageTableSubscriber.start();
|
||||
postgresSubscribableChannel.subscribe(message -> {
|
||||
payloads.add(message.getPayload());
|
||||
@@ -326,7 +339,7 @@ public class PostgresChannelMessageTableSubscriberTests implements PostgresConta
|
||||
|
||||
private static class ConnectionSupplier implements PgConnectionSupplier {
|
||||
|
||||
Runnable onGetConnection;
|
||||
Consumer<PgConnection> onGetConnection;
|
||||
|
||||
@Override
|
||||
public PgConnection get() throws SQLException {
|
||||
@@ -335,10 +348,11 @@ public class PostgresChannelMessageTableSubscriberTests implements PostgresConta
|
||||
POSTGRES_CONTAINER.getPassword())
|
||||
.unwrap(PgConnection.class);
|
||||
if (this.onGetConnection != null) {
|
||||
this.onGetConnection.run();
|
||||
this.onGetConnection.accept(conn);
|
||||
}
|
||||
return conn;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user