Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

PostgresChannelMessageTableSubscriber: Renew connection only if invalid

Browse Source 
Fixes: #9111

An evolution of the #9061: renew the connection only when we need to.

(cherry picked from commit da29e2da6a)

# Conflicts:
#	spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java
This commit is contained in:
Johannes Edmeier
2024-05-03 14:55:47 +02:00
committed by Artem Bilan
parent cb5d20e8eb
commit 00c75acd66
2 changed files with 88 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

128
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java
View File

@@ -169,70 +169,84 @@ public final class PostgresChannelMessageTableSubscriber implements SmartLifecyc
this.latch = new CountDownLatch(1);
CountDownLatch startingLatch = new CountDownLatch(1);
this.future = executorToUse.submit(() -> {
try {
while (isActive()) {
try {
PgConnection conn = this.connectionSupplier.get();
try (Statement stmt = conn.createStatement()) {
stmt.execute("LISTEN " + this.tablePrefix.toLowerCase() + "channel_message_notify");
}
catch (Exception ex) {
try {
conn.close();
}
catch (Exception suppressed) {
ex.addSuppressed(suppressed);
}
throw ex;
}
this.subscriptionsMap.values()
.forEach(subscriptions -> subscriptions.forEach(Subscription::notifyUpdate));
try {
this.connection = conn;
while (isActive()) {
startingLatch.countDown();
this.future = executorToUse.submit(() -> doStart(startingLatch));
PGNotification[] notifications = conn.getNotifications((int) this.notificationTimeout.toMillis());
// Unfortunately, there is no good way of interrupting a notification
// poll but by closing its connection.
if (!isActive()) {
return;
}
if (notifications == null || notifications.length == 0) {
//We did not receive any notifications within the timeout period.
//We will close the connection and re-establish it.
break;
}
for (PGNotification notification : notifications) {
String parameter = notification.getParameter();
Set<Subscription> subscriptions = this.subscriptionsMap.get(parameter);
if (subscriptions == null) {
continue;
}
for (Subscription subscription : subscriptions) {
subscription.notifyUpdate();
}
}
}
}
finally {
try {
if (!startingLatch.await(5, TimeUnit.SECONDS)) {
throw new IllegalStateException("Failed to start " + this);
}
}
catch (InterruptedException ex) {
Thread.currentThread().interrupt();
throw new IllegalStateException("Failed to start " + this, ex);
}
}
private void doStart(CountDownLatch startingLatch) {
try {
while (isActive()) {
try {
PgConnection conn = this.connectionSupplier.get();
try (Statement stmt = conn.createStatement()) {
stmt.execute("LISTEN " + this.tablePrefix.toLowerCase() + "channel_message_notify");
}
catch (Exception ex) {
try {
conn.close();
}
}
catch (Exception e) {
// The getNotifications method does not throw a meaningful message on interruption.
// Therefore, we do not log an error, unless it occurred while active.
if (isActive()) {
LOGGER.error(e, "Failed to poll notifications from Postgres database");
catch (Exception suppressed) {
ex.addSuppressed(suppressed);
}
throw ex;
}
this.subscriptionsMap.values()
.forEach(subscriptions -> subscriptions.forEach(Subscription::notifyUpdate));
try {
this.connection = conn;
while (isActive()) {
startingLatch.countDown();
PGNotification[] notifications = conn.getNotifications((int) this.notificationTimeout.toMillis());
// Unfortunately, there is no good way of interrupting a notification
// poll but by closing its connection.
if (!isActive()) {
return;
}
if ((notifications == null || notifications.length == 0) && !conn.isValid(1)) {
//We did not receive any notifications within the timeout period.
//If the connection is still valid, we will continue polling
//Otherwise, we will close the connection and re-establish it.
break;
}
for (PGNotification notification : notifications) {
String parameter = notification.getParameter();
Set<Subscription> subscriptions = this.subscriptionsMap.get(parameter);
if (subscriptions == null) {
continue;
}
for (Subscription subscription : subscriptions) {
subscription.notifyUpdate();
}
}
}
}
finally {
conn.close();
}
}
catch (Exception e) {
// The getNotifications method does not throw a meaningful message on interruption.
// Therefore, we do not log an error, unless it occurred while active.
if (isActive()) {
LOGGER.error(e, "Failed to poll notifications from Postgres database");
}
}
}
finally {
this.latch.countDown();
}
});
}
finally {
this.latch.countDown();
}
try {
if (!startingLatch.await(5, TimeUnit.SECONDS)) {

20
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriberTests.java
View File

@@ -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;
@@ -284,7 +286,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());
@@ -340,7 +353,7 @@ public class PostgresChannelMessageTableSubscriberTests implements PostgresConta
private static class ConnectionSupplier implements PgConnectionSupplier {
Runnable onGetConnection;
Consumer<PgConnection> onGetConnection;
@Override
public PgConnection get() throws SQLException {
@@ -349,10 +362,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;
}
}
}