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 b42e949c18..e7ede05435 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 @@ -13,9 +13,9 @@ package org.springframework.integration.aggregator; -import java.util.ArrayList; import java.util.Collection; import java.util.Collections; +import java.util.Comparator; import java.util.Date; import java.util.HashMap; import java.util.List; @@ -81,6 +81,7 @@ import org.springframework.util.CollectionUtils; * @author Gary Russell * @author Artem Bilan * @author David Liu + * @author Enrique Rodríguez * @since 2.0 */ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageProducingHandler @@ -90,6 +91,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP public static final long DEFAULT_SEND_TIMEOUT = 1000L; + private final Comparator> sequenceNumberComparator = new SequenceNumberComparator(); + private final Map> expireGroupScheduledFutures = new HashMap>(); protected volatile MessageGroupStore messageStore; @@ -588,11 +591,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } protected int findLastReleasedSequenceNumber(Object groupId, Collection> partialSequence) { - List> sorted = new ArrayList>(partialSequence); - Collections.sort(sorted, new SequenceNumberComparator()); - - Message lastReleasedMessage = sorted.get(partialSequence.size() - 1); - + Message lastReleasedMessage = Collections.max(partialSequence, this.sequenceNumberComparator); return new IntegrationMessageHeaderAccessor(lastReleasedMessage).getSequenceNumber(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java index 2372392d21..29c01d37cf 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java @@ -16,11 +16,9 @@ package org.springframework.integration.aggregator; -import java.util.ArrayList; import java.util.Collection; import java.util.Collections; import java.util.Comparator; -import java.util.List; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -38,12 +36,13 @@ import org.springframework.messaging.Message; * @author Dave Syer * @author Iwein Fuld * @author Oleg Zhurakousky + * @author Enrique Rodríguez */ public class SequenceSizeReleaseStrategy implements ReleaseStrategy { private static final Log logger = LogFactory.getLog(SequenceSizeReleaseStrategy.class); - private volatile Comparator> comparator = new SequenceNumberComparator(); + private final Comparator> comparator = new SequenceNumberComparator(); private volatile boolean releasePartialSequences; @@ -77,10 +76,9 @@ public class SequenceSizeReleaseStrategy implements ReleaseStrategy { if (logger.isTraceEnabled()) { logger.trace("Considering partial release of group [" + messageGroup + "]"); } - List> sorted = new ArrayList>(messages); - Collections.sort(sorted, comparator); + Message minMessage = Collections.min(messages, this.comparator); - int nextSequenceNumber = new IntegrationMessageHeaderAccessor(sorted.get(0)).getSequenceNumber(); + int nextSequenceNumber = new IntegrationMessageHeaderAccessor(minMessage).getSequenceNumber(); int lastReleasedMessageSequence = messageGroup.getLastReleasedMessageSequenceNumber(); if (nextSequenceNumber - lastReleasedMessageSequence == 1){ diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java index 222f830dbb..600763ab95 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java @@ -17,6 +17,7 @@ package org.springframework.integration.channel; import java.util.Collections; +import java.util.Comparator; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; @@ -50,12 +51,14 @@ import org.springframework.util.StringUtils; public abstract class AbstractMessageChannel extends IntegrationObjectSupport implements MessageChannel, TrackableComponent, ChannelInterceptorAware { + private final ChannelInterceptorList interceptors = new ChannelInterceptorList(); + + private final Comparator orderComparator = new OrderComparator(); + private volatile boolean shouldTrack = false; private volatile Class[] datatypes = new Class[0]; - private final ChannelInterceptorList interceptors = new ChannelInterceptorList(); - private volatile String fullChannelName; private volatile MessageConverter messageConverter; @@ -96,7 +99,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport */ @Override public void setInterceptors(List interceptors) { - Collections.sort(interceptors, new OrderComparator()); + Collections.sort(interceptors, this.orderComparator); this.interceptors.set(interceptors); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/GlobalChannelInterceptorProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/GlobalChannelInterceptorProcessor.java index 9410aa0dcf..b413144152 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/GlobalChannelInterceptorProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/GlobalChannelInterceptorProcessor.java @@ -153,7 +153,7 @@ final class GlobalChannelInterceptorProcessor implements BeanFactoryAware, Smart tempInterceptors.add(globalChannelInterceptorWrapper); } } - Collections.sort(tempInterceptors, comparator); + Collections.sort(tempInterceptors, this.comparator); if (!tempInterceptors.isEmpty()) { for (int i = tempInterceptors.size() - 1; i >= 0; i--) { ChannelInterceptor channelInterceptor = tempInterceptors.get(i).getChannelInterceptor(); diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelPriorityMessageStore.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelPriorityMessageStore.java index d472cbfeb9..aae4e74c55 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelPriorityMessageStore.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelPriorityMessageStore.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.redis.store; import java.util.Collections; @@ -45,6 +46,15 @@ import org.springframework.util.Assert; */ public class RedisChannelPriorityMessageStore extends RedisChannelMessageStore implements PriorityCapableChannelMessageStore { + private final Comparator keysComparator = new Comparator() { + + @Override + public int compare(String s1, String s2) { + return s2.compareTo(s1); + } + + }; + public RedisChannelPriorityMessageStore(RedisConnectionFactory connectionFactory) { super(connectionFactory); } @@ -111,13 +121,7 @@ public class RedisChannelPriorityMessageStore extends RedisChannelMessageStore i Assert.isInstanceOf(String.class, key); list.add((String) key); } - Collections.sort(list, new Comparator() { - - @Override - public int compare(String s1, String s2) { - return s2.compareTo(s1); - } - }); + Collections.sort(list, this.keysComparator); return list; }