diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index cde44307ce..8413b3314d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java @@ -63,9 +63,9 @@ import org.springframework.util.CollectionUtils; /** * Abstract Message handler that holds a buffer of correlated messages in a * {@link MessageStore}. This class takes care of correlated groups of messages - * that can be completed in batches. It is useful for custom implementation of MessageHandlers that require correlation - * and is used as a base class for Aggregator - {@link AggregatingMessageHandler} and - * Resequencer - {@link ResequencingMessageHandler}, + * that can be completed in batches. It is useful for custom implementation of + * MessageHandlers that require correlation and is used as a base class for Aggregator - + * {@link AggregatingMessageHandler} and Resequencer - {@link ResequencingMessageHandler}, * or custom implementations requiring correlation. *

* To customize this handler inject {@link CorrelationStrategy}, @@ -94,10 +94,10 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP private final Map> expireGroupScheduledFutures = new HashMap>(); - protected volatile MessageGroupStore messageStore; - private final MessageGroupProcessor outputProcessor; + private volatile MessageGroupStore messageStore; + private volatile CorrelationStrategy correlationStrategy; private volatile ReleaseStrategy releaseStrategy; @@ -486,8 +486,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } private void discardMessage(Message message) { - MessageChannel discardChannel = getDiscardChannel(); - this.messagingTemplate.send(discardChannel, message); + this.messagingTemplate.send(getDiscardChannel(), message); } /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java index 1ee5405a57..0a53cd5c5b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java @@ -64,17 +64,18 @@ public class AggregatingMessageHandler extends AbstractCorrelatingMessageHandler @Override protected void afterRelease(MessageGroup messageGroup, Collection> completedMessages) { Object groupId = messageGroup.getGroupId(); - this.messageStore.completeGroup(groupId); + MessageGroupStore messageStore = getMessageStore(); + messageStore.completeGroup(groupId); if (this.expireGroupsUponCompletion) { remove(messageGroup); } else { - if (this.messageStore instanceof SimpleMessageStore) { - ((SimpleMessageStore) this.messageStore).clearMessageGroup(groupId); + if (messageStore instanceof SimpleMessageStore) { + ((SimpleMessageStore) messageStore).clearMessageGroup(groupId); } else { - this.messageStore.removeMessagesFromGroup(groupId, messageGroup.getMessages()); + messageStore.removeMessagesFromGroup(groupId, messageGroup.getMessages()); } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageHandler.java index 8987eea6b0..29b2356f39 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageHandler.java @@ -82,19 +82,20 @@ public class ResequencingMessageHandler extends AbstractCorrelatingMessageHandle } else { Object groupId = messageGroup.getGroupId(); + MessageGroupStore messageStore = getMessageStore(); if (completedMessages != null) { int lastReleasedSequenceNumber = findLastReleasedSequenceNumber(groupId, completedMessages); - this.messageStore.setLastReleasedSequenceNumberForGroup(groupId, lastReleasedSequenceNumber); - if (this.messageStore instanceof SimpleMessageStore + messageStore.setLastReleasedSequenceNumberForGroup(groupId, lastReleasedSequenceNumber); + if (messageStore instanceof SimpleMessageStore && completedMessages.size() == messageGroup.size()) { - ((SimpleMessageStore) this.messageStore).clearMessageGroup(groupId); + ((SimpleMessageStore) messageStore).clearMessageGroup(groupId); } else { - this.messageStore.removeMessagesFromGroup(groupId, completedMessages); + messageStore.removeMessagesFromGroup(groupId, completedMessages); } } if (timeout) { - this.messageStore.completeGroup(groupId); + messageStore.completeGroup(groupId); } } } diff --git a/spring-integration-groovy/src/main/java/org/springframework/integration/groovy/GroovyScriptExecutingMessageProcessor.java b/spring-integration-groovy/src/main/java/org/springframework/integration/groovy/GroovyScriptExecutingMessageProcessor.java index b3ed7e045f..60a5e73a51 100644 --- a/spring-integration-groovy/src/main/java/org/springframework/integration/groovy/GroovyScriptExecutingMessageProcessor.java +++ b/spring-integration-groovy/src/main/java/org/springframework/integration/groovy/GroovyScriptExecutingMessageProcessor.java @@ -16,6 +16,8 @@ package org.springframework.integration.groovy; +import java.security.AccessController; +import java.security.PrivilegedAction; import java.util.Map; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; @@ -66,7 +68,14 @@ public class GroovyScriptExecutingMessageProcessor extends AbstractScriptExecuti private volatile ScriptSource scriptSource; - private volatile GroovyClassLoader groovyClassLoader = new GroovyClassLoader(ClassUtils.getDefaultClassLoader()); + private volatile GroovyClassLoader groovyClassLoader = AccessController.doPrivileged( + new PrivilegedAction() { + + public GroovyClassLoader run() { + return new GroovyClassLoader(ClassUtils.getDefaultClassLoader()); + } + + }); private volatile Class scriptClass; diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/DefaultLockRepository.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/DefaultLockRepository.java index 5b545ada0f..1bacca6cf3 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/DefaultLockRepository.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/DefaultLockRepository.java @@ -52,11 +52,16 @@ public class DefaultLockRepository implements LockRepository, InitializingBean { */ public static final String DEFAULT_TABLE_PREFIX = "INT_"; + /** + * Default value for the time-to-live property. + */ + public static final int DEFAULT_TTL = 10000; + private final String id = UUID.randomUUID().toString(); private final JdbcTemplate template; - private int ttl = 10000; + private int ttl = DEFAULT_TTL; private String prefix = DEFAULT_TABLE_PREFIX; @@ -108,7 +113,7 @@ public class DefaultLockRepository implements LockRepository, InitializingBean { } @Override - public void afterPropertiesSet() throws Exception { + public void afterPropertiesSet() { this.deleteQuery = String.format(this.deleteQuery, this.prefix); this.deleteExpiredQuery = String.format(this.deleteExpiredQuery, this.prefix); this.deleteAllQuery = String.format(this.deleteAllQuery, this.prefix); diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java index bceae2fab5..c66ebab9b4 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java @@ -25,6 +25,8 @@ import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; +import org.springframework.dao.CannotAcquireLockException; +import org.springframework.dao.DataAccessResourceFailureException; import org.springframework.integration.support.locks.DefaultLockRegistry; import org.springframework.integration.support.locks.ExpirableLockRegistry; import org.springframework.integration.support.locks.LockRegistry; @@ -111,58 +113,59 @@ public class JdbcLockRegistry implements ExpirableLockRegistry { @Override public void lock() { this.delegate.lock(); - try { - while (true) { - try { - while (!doLock()) { - Thread.sleep(100); - } - break; + while (true) { + try { + while (!doLock()) { + Thread.sleep(100); //NOSONAR } - catch (TransactionTimedOutException e) { - // try again - } - catch (InterruptedException e) { + break; + } + catch (TransactionTimedOutException e) { + // try again + } + catch (InterruptedException e) { /* * This method must be uninterruptible so catch and ignore * interrupts and only break out of the while loop when * we get the lock. */ - } + } + catch (Exception e) { + this.delegate.unlock(); + rethrowAsLockException(e); } } - catch (Exception e) { - this.delegate.unlock(); - throw new RuntimeException("Failed to lock mutex at " + this.path, e); - } + } + + private void rethrowAsLockException(Exception e) { + throw new CannotAcquireLockException("Failed to lock mutex at " + this.path, e); } @Override public void lockInterruptibly() throws InterruptedException { this.delegate.lockInterruptibly(); - try { - while (true) { - try { - while (!this.doLock()) { - Thread.sleep(100); - if (Thread.currentThread().isInterrupted()) { - throw new InterruptedException(); - } + while (true) { + try { + while (!doLock()) { + Thread.sleep(100); //NOSONAR + if (Thread.currentThread().isInterrupted()) { + throw new InterruptedException(); } - break; - } - catch (TransactionTimedOutException e) { - // try again } + break; + } + catch (TransactionTimedOutException e) { + // try again + } + catch (InterruptedException ie) { + this.delegate.unlock(); + Thread.currentThread().interrupt(); + throw ie; + } + catch (Exception e) { + this.delegate.unlock(); + rethrowAsLockException(e); } - } - catch (InterruptedException ie) { - this.delegate.unlock(); - throw ie; - } - catch (Exception e) { - this.delegate.unlock(); - throw new RuntimeException("Failed to lock mutex at " + this.path, e); } } @@ -183,27 +186,25 @@ public class JdbcLockRegistry implements ExpirableLockRegistry { if (!this.delegate.tryLock(time, unit)) { return false; } - try { - long expire = now + TimeUnit.MILLISECONDS.convert(time, unit); - boolean acquired; - while (true) { - try { - while (!(acquired = doLock()) && System.currentTimeMillis() < expire) { - Thread.sleep(100); - } - if (!acquired) { - this.delegate.unlock(); - } - return acquired; + long expire = now + TimeUnit.MILLISECONDS.convert(time, unit); + boolean acquired; + while (true) { + try { + while (!(acquired = doLock()) && System.currentTimeMillis() < expire) { //NOSONAR + Thread.sleep(100); //NOSONAR } - catch (TransactionTimedOutException e) { - // try again + if (!acquired) { + this.delegate.unlock(); } + return acquired; + } + catch (TransactionTimedOutException e) { + // try again + } + catch (Exception e) { + this.delegate.unlock(); + rethrowAsLockException(e); } - } - catch (Exception e) { - this.delegate.unlock(); - throw new RuntimeException("Failed to lock mutex at " + this.path, e); } } @@ -228,7 +229,7 @@ public class JdbcLockRegistry implements ExpirableLockRegistry { this.mutex.delete(this.path); } catch (Exception e) { - throw new RuntimeException("Failed to release mutex at " + this.path, e); + throw new DataAccessResourceFailureException("Failed to release mutex at " + this.path, e); } finally { this.delegate.unlock(); diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java index c13b262cfc..aa53255760 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java @@ -36,6 +36,7 @@ import java.util.concurrent.locks.Lock; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.dao.CannotAcquireLockException; import org.springframework.dao.DataAccessException; import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.RedisConnectionFactory; @@ -80,6 +81,7 @@ import org.springframework.util.Assert; * * @author Gary Russell * @author Konstantin Yakimov + * @author Artem Bilan * @since 4.0 * */ @@ -155,7 +157,7 @@ public final class RedisLockRegistry implements LockRegistry { this.redisTemplate = new RedisTemplate(); this.redisTemplate.setConnectionFactory(connectionFactory); this.redisTemplate.setKeySerializer(new StringRedisSerializer()); - this.redisTemplate.setValueSerializer(new LockSerializer()); + this.redisTemplate.setValueSerializer(this.lockSerializer); this.redisTemplate.afterPropertiesSet(); this.registryKey = registryKey; this.expireAfter = expireAfter; @@ -323,27 +325,29 @@ public final class RedisLockRegistry implements LockRegistry { public void lock() { Lock localLock = RedisLockRegistry.this.localRegistry.obtain(this.lockKey); localLock.lock(); - try { - while (true) { - try { - while (!this.obtainLock()) { - Thread.sleep(100); - } - break; + while (true) { + try { + while (!this.obtainLock()) { + Thread.sleep(100); //NOSONAR } - catch (InterruptedException e) { + break; + } + catch (InterruptedException e) { /* * This method must be uninterruptible so catch and ignore * interrupts and only break out of the while loop when * we get the lock. */ - } + } + catch (Exception e) { + localLock.unlock(); + rethrowAsLockException(e); } } - catch (Exception e) { - localLock.unlock(); - throw new RuntimeException(e); - } + } + + private void rethrowAsLockException(Exception e) { + throw new CannotAcquireLockException("Failed to lock mutex at " + this.lockKey, e); } @Override @@ -352,16 +356,17 @@ public final class RedisLockRegistry implements LockRegistry { localLock.lockInterruptibly(); try { while (!this.obtainLock()) { - Thread.sleep(100); + Thread.sleep(100); //NOSONAR } } catch (InterruptedException ie) { localLock.unlock(); + Thread.currentThread().interrupt(); throw ie; } catch (Exception e) { localLock.unlock(); - throw new RuntimeException(e); + rethrowAsLockException(e); } } @@ -380,8 +385,9 @@ public final class RedisLockRegistry implements LockRegistry { } catch (Exception e) { localLock.unlock(); - throw new RuntimeException(e); + rethrowAsLockException(e); } + return false; } private boolean obtainLock() { @@ -455,9 +461,9 @@ public final class RedisLockRegistry implements LockRegistry { } try { long expire = System.currentTimeMillis() + TimeUnit.MILLISECONDS.convert(time, unit); - boolean acquired = false; - while (!(acquired = this.obtainLock()) && System.currentTimeMillis() < expire) { - Thread.sleep(100); + boolean acquired; + while (!(acquired = obtainLock()) && System.currentTimeMillis() < expire) { //NOSONAR + Thread.sleep(100); //NOSONAR } if (!acquired) { localLock.unlock(); @@ -466,8 +472,9 @@ public final class RedisLockRegistry implements LockRegistry { } catch (Exception e) { localLock.unlock(); - throw new RuntimeException(e); + rethrowAsLockException(e); } + return false; } @Override