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

More Sonar fixes

Browse Source
This commit is contained in:
Artem Bilan
2022-11-03 17:28:54 -04:00
parent 7084e9654a
commit d31f309752
15 changed files with 76 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -66,7 +66,7 @@ public final class PostgresChannelMessageTableSubscriber implements SmartLifecyc
private static final LogAccessor LOGGER = new LogAccessor(PostgresChannelMessageTableSubscriber.class);
private final Map<String, Set<Subscription>> subscriptions = new ConcurrentHashMap<>();
private final Map<String, Set<Subscription>> subscriptionsMap = new ConcurrentHashMap<>();
private final PgConnectionSupplier connectionSupplier;
@@ -120,7 +120,7 @@ public final class PostgresChannelMessageTableSubscriber implements SmartLifecyc
public boolean subscribe(Subscription subscription) {
String subscriptionKey = subscription.getRegion() + " " + getKey(subscription.getGroupId());
Set<Subscription> subscriptions =
this.subscriptions.computeIfAbsent(subscriptionKey, __ -> ConcurrentHashMap.newKeySet());
this.subscriptionsMap.computeIfAbsent(subscriptionKey, __ -> ConcurrentHashMap.newKeySet());
return subscriptions.add(subscription);
}
@@ -131,7 +131,7 @@ public final class PostgresChannelMessageTableSubscriber implements SmartLifecyc
*/
public boolean unsubscribe(Subscription subscription) {
String subscriptionKey = subscription.getRegion() + " " + getKey(subscription.getGroupId());
Set<Subscription> subscriptions = this.subscriptions.get(subscriptionKey);
Set<Subscription> subscriptions = this.subscriptionsMap.get(subscriptionKey);
return subscriptions != null && subscriptions.remove(subscription);
}
@@ -140,16 +140,16 @@ public final class PostgresChannelMessageTableSubscriber implements SmartLifecyc
if (this.latch.getCount() > 0) {
return;
}
ExecutorService executor = this.executor;
if (executor == null) {
ExecutorService executorToUse = this.executor;
if (executorToUse == null) {
CustomizableThreadFactory threadFactory =
new CustomizableThreadFactory("postgres-channel-message-table-subscriber-");
threadFactory.setDaemon(true);
executor = Executors.newSingleThreadExecutor(threadFactory);
this.executor = executor;
executorToUse = Executors.newSingleThreadExecutor(threadFactory);
this.executor = executorToUse;
}
this.latch = new CountDownLatch(1);
this.future = executor.submit(() -> {
this.future = executorToUse.submit(() -> {
try {
while (isActive()) {
try {
@@ -157,16 +157,16 @@ public final class PostgresChannelMessageTableSubscriber implements SmartLifecyc
try (Statement stmt = conn.createStatement()) {
stmt.execute("LISTEN " + this.tablePrefix.toLowerCase() + "channel_message_notify");
}
catch (Throwable t) {
catch (Exception ex) {
try {
conn.close();
}
catch (Throwable suppressed) {
t.addSuppressed(suppressed);
catch (Exception suppressed) {
ex.addSuppressed(suppressed);
}
throw t;
throw ex;
}
this.subscriptions.values()
this.subscriptionsMap.values()
.forEach(subscriptions -> subscriptions.forEach(Subscription::notifyUpdate));
try {
this.connection = conn;
@@ -180,7 +180,7 @@ public final class PostgresChannelMessageTableSubscriber implements SmartLifecyc
if (notifications != null) {
for (PGNotification notification : notifications) {
String parameter = notification.getParameter();
Set<Subscription> subscriptions = this.subscriptions.get(parameter);
Set<Subscription> subscriptions = this.subscriptionsMap.get(parameter);
if (subscriptions == null) {
continue;
}
@@ -202,10 +202,6 @@ public final class PostgresChannelMessageTableSubscriber implements SmartLifecyc
LOGGER.error(e, "Failed to poll notifications from Postgres database");
}
}
catch (Throwable t) {
LOGGER.error(t, "Failed to poll notifications from Postgres database");
return;
}
}
}
finally {
@@ -224,11 +220,10 @@ public final class PostgresChannelMessageTableSubscriber implements SmartLifecyc
@Override
public synchronized void stop() {
Future<?> future = this.future;
if (future.isDone()) {
if (this.future.isDone()) {
return;
}
future.cancel(true);
this.future.cancel(true);
PgConnection conn = this.connection;
if (conn != null) {
try {

14
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcMessageStore.java
View File

@@ -16,7 +16,6 @@
package org.springframework.integration.jdbc.store;
import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
@@ -50,9 +49,6 @@ import org.springframework.jdbc.core.JdbcOperations;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.core.SingleColumnRowMapper;
import org.springframework.jdbc.support.JdbcAccessor;
import org.springframework.jdbc.support.JdbcUtils;
import org.springframework.jdbc.support.MetaDataAccessException;
import org.springframework.jdbc.support.lob.DefaultLobHandler;
import org.springframework.jdbc.support.lob.LobHandler;
import org.springframework.jmx.export.annotation.ManagedAttribute;
@@ -171,8 +167,6 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
private final JdbcOperations jdbcTemplate;
private final String vendorName;
private final Map<Query, String> queryCache = new ConcurrentHashMap<>();
private String region = "DEFAULT";
@@ -205,14 +199,6 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
Assert.notNull(jdbcOperations, "'dataSource' must not be null");
this.jdbcTemplate = jdbcOperations;
this.serializer = new SerializingConverter();
try {
this.vendorName =
JdbcUtils.extractDatabaseMetaData(((JdbcAccessor) jdbcOperations).getDataSource(), // NOSONAR
DatabaseMetaData::getDatabaseProductName);
}
catch (MetaDataAccessException ex) {
throw new IllegalStateException("Cannot extract database vendor name", ex);
}
}
@Override