Merge branch 'INT-3528'

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java

@@ -13,12 +13,11 @@
 
 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;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.ScheduledFuture;
@@ -81,6 +80,7 @@
  * @author Gary Russell
  * @author Artem Bilan
  * @author David Liu
+ * @author Enrique Rodríguez
  * @since 2.0
  */
 public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageProducingHandler
@@ -90,6 +90,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 
 	public static final long DEFAULT_SEND_TIMEOUT = 1000L;
 
+	private final Comparator<Message<?>> sequenceNumberComparator = new SequenceNumberComparator();
+
 	private final Map<UUID, ScheduledFuture<?>> expireGroupScheduledFutures = new HashMap<UUID, ScheduledFuture<?>>();
 
 	protected volatile MessageGroupStore messageStore;
@@ -588,11 +590,7 @@ void remove(MessageGroup group) {
 	}
 
 	protected int findLastReleasedSequenceNumber(Object groupId, Collection<Message<?>> partialSequence) {
-		List<Message<?>> sorted = new ArrayList<Message<?>>(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();
 	}
 

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 @@
  * @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<Message<?>> comparator = new SequenceNumberComparator();
+	private final Comparator<Message<?>> comparator = new SequenceNumberComparator();
 
 	private volatile boolean releasePartialSequences;
 
@@ -77,10 +76,9 @@ public boolean canRelease(MessageGroup messageGroup) {
 			if (logger.isTraceEnabled()) {
 				logger.trace("Considering partial release of group [" + messageGroup + "]");
 			}
-			List<Message<?>> sorted = new ArrayList<Message<?>>(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){

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 @@
 public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 		implements MessageChannel, TrackableComponent, ChannelInterceptorAware {
 
+	private final ChannelInterceptorList interceptors = new ChannelInterceptorList();
+
+	private final Comparator<Object> 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 void setDatatypes(Class<?>... datatypes) {
 	 */
 	@Override
 	public void setInterceptors(List<ChannelInterceptor> interceptors) {
-		Collections.sort(interceptors, new OrderComparator());
+		Collections.sort(interceptors, this.orderComparator);
 		this.interceptors.set(interceptors);
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/config/GlobalChannelInterceptorProcessor.java

@@ -153,7 +153,7 @@ private void addMatchingInterceptors(ChannelInterceptorAware channel, String bea
 				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();

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 @@
  */
 public class RedisChannelPriorityMessageStore extends RedisChannelMessageStore implements PriorityCapableChannelMessageStore {
 
+	private final Comparator<String> keysComparator = new Comparator<String>() {
+
+		@Override
+		public int compare(String s1, String s2) {
+			return s2.compareTo(s1);
+		}
+
+	};
+
 	public RedisChannelPriorityMessageStore(RedisConnectionFactory connectionFactory) {
 		super(connectionFactory);
 	}
@@ -111,13 +121,7 @@ private List<String> sortedKeys(String groupId) {
 			Assert.isInstanceOf(String.class, key);
 			list.add((String) key);
 		}
-		Collections.sort(list, new Comparator<String>() {
-
-			@Override
-			public int compare(String s1, String s2) {
-				return s2.compareTo(s1);
-			}
-		});
+		Collections.sort(list, this.keysComparator);
 		return list;
 	}