diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java index ceadd23047..b7a6787873 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -20,8 +20,10 @@ import java.util.ArrayList; import java.util.Collection; import java.util.HashSet; import java.util.Iterator; +import java.util.List; import java.util.Set; import java.util.UUID; +import java.util.stream.Collectors; import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.messaging.Message; @@ -243,11 +245,21 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS if (mgm != null) { Assert.isInstanceOf(MessageGroupMetadata.class, mgm); MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm; - for (Message messageToRemove : messages) { - UUID messageId = messageToRemove.getHeaders().getId(); - messageGroupMetadata.remove(messageId); - doRemove(this.messagePrefix + messageId); - } + + List ids = + messages.stream() + .map(messageToRemove -> messageToRemove.getHeaders().getId()) + .collect(Collectors.toList()); + + messageGroupMetadata.removeAll(ids); + + List messageIds = + ids.stream() + .map(id -> this.messagePrefix + id) + .collect(Collectors.toList()); + + doRemoveAll(messageIds); + messageGroupMetadata.setLastModified(System.currentTimeMillis()); doStore(this.groupPrefix + groupId, messageGroupMetadata); } @@ -275,10 +287,13 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS Assert.isInstanceOf(MessageGroupMetadata.class, mgm); MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm; - Iterator messageIds = messageGroupMetadata.messageIdIterator(); - while (messageIds.hasNext()) { - removeMessage(messageIds.next()); - } + List messageIds = + messageGroupMetadata.getMessageIds() + .stream() + .map(id -> this.messagePrefix + id) + .collect(Collectors.toList()); + + doRemoveAll(messageIds); } } @@ -325,7 +340,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS @Override public Collection> getMessagesForGroup(Object groupId) { MessageGroupMetadata groupMetadata = getGroupMetadata(groupId); - ArrayList> messages = new ArrayList>(); + ArrayList> messages = new ArrayList<>(); if (groupMetadata != null) { Iterator messageIds = groupMetadata.messageIdIterator(); while (messageIds.hasNext()) { @@ -345,7 +360,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS } private Collection normalizeKeys(Collection keys) { - Set normalizedKeys = new HashSet(); + Set normalizedKeys = new HashSet<>(); for (Object key : keys) { String strKey = (String) key; if (strKey.startsWith(this.groupPrefix)) { @@ -378,6 +393,8 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS protected abstract Object doRemove(Object id); + protected abstract void doRemoveAll(Collection ids); + protected abstract Collection doListKeys(String keyPattern); private final class MessageGroupIterator implements Iterator { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java index 1930b98a6b..26551eaa55 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java @@ -17,6 +17,7 @@ package org.springframework.integration.store; import java.io.Serializable; +import java.util.Collection; import java.util.Iterator; import java.util.LinkedList; import java.util.List; @@ -39,7 +40,7 @@ public class MessageGroupMetadata implements Serializable { private static final long serialVersionUID = 1L; - private List messageIds = new LinkedList(); + private List messageIds = new LinkedList<>(); private long timestamp; @@ -68,6 +69,10 @@ public class MessageGroupMetadata implements Serializable { this.messageIds.remove(messageId); } + public void removeAll(Collection messageIds) { + this.messageIds.removeAll(messageIds); + } + boolean add(UUID messageId) { return !this.messageIds.contains(messageId) && this.messageIds.add(messageId); } diff --git a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/GemfireMessageStore.java b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/GemfireMessageStore.java index 114c61fde3..46d1022042 100644 --- a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/GemfireMessageStore.java +++ b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/GemfireMessageStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -96,6 +96,11 @@ public class GemfireMessageStore extends AbstractKeyValueMessageStore { return this.messageStoreRegion.remove(id); } + @Override + protected void doRemoveAll(Collection ids) { + this.messageStoreRegion.removeAll(ids); + } + @Override protected Collection doListKeys(String keyPattern) { Assert.hasText(keyPattern, "'keyPattern' must not be empty"); diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageStore.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageStore.java index 2402172975..431432fa93 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageStore.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2007-2017 the original author or authors. + * Copyright 2007-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -26,6 +26,7 @@ import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer import org.springframework.data.redis.serializer.RedisSerializer; import org.springframework.data.redis.serializer.SerializationException; import org.springframework.data.redis.serializer.StringRedisSerializer; +import org.springframework.integration.redis.util.RedisUtils; import org.springframework.integration.store.AbstractKeyValueMessageStore; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.MessageStore; @@ -66,7 +67,7 @@ public class RedisMessageStore extends AbstractKeyValueMessageStore implements B */ public RedisMessageStore(RedisConnectionFactory connectionFactory, String prefix) { super(prefix); - this.redisTemplate = new RedisTemplate(); + this.redisTemplate = new RedisTemplate<>(); this.redisTemplate.setConnectionFactory(connectionFactory); this.redisTemplate.setKeySerializer(new StringRedisSerializer()); this.redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer()); @@ -130,11 +131,26 @@ public class RedisMessageStore extends AbstractKeyValueMessageStore implements B Assert.notNull(id, "'id' must not be null"); Object removedObject = this.doRetrieve(id); if (removedObject != null) { - this.redisTemplate.delete(id); + if (RedisUtils.isUnlinkAvailable(this.redisTemplate)) { + this.redisTemplate.unlink(id); + } + else { + this.redisTemplate.delete(id); + } } return removedObject; } + @Override + protected void doRemoveAll(Collection ids) { + if (RedisUtils.isUnlinkAvailable(this.redisTemplate)) { + this.redisTemplate.unlink(ids); + } + else { + this.redisTemplate.delete(ids); + } + } + @Override protected Collection doListKeys(String keyPattern) { Assert.hasText(keyPattern, "'keyPattern' must not be empty"); 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 c3be6e9c0d..d2e2c4aabe 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 @@ -302,11 +302,10 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl } try { if (Thread.currentThread().isInterrupted()) { - RedisLockRegistry.this.executor.execute(() -> - RedisLockRegistry.this.redisTemplate.delete(this.lockKey)); + RedisLockRegistry.this.executor.execute(this::removeLockKey); } else { - RedisLockRegistry.this.redisTemplate.delete(this.lockKey); + removeLockKey(); } if (logger.isDebugEnabled()) { @@ -321,6 +320,15 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl } } + private void removeLockKey() { + if (RedisUtils.isUnlinkAvailable(RedisLockRegistry.this.redisTemplate)) { + RedisLockRegistry.this.redisTemplate.unlink(this.lockKey); + } + else { + RedisLockRegistry.this.redisTemplate.delete(this.lockKey); + } + } + @Override public Condition newCondition() { throw new UnsupportedOperationException("Conditions are not supported"); diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisUtils.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisUtils.java new file mode 100644 index 0000000000..a5559e23df --- /dev/null +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisUtils.java @@ -0,0 +1,66 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.redis.util; + +import java.util.Properties; + +import org.springframework.data.redis.core.RedisCallback; +import org.springframework.data.redis.core.RedisOperations; + +/** + * A set of utility methods for common Redis functions. + * + * @author Artem Bilan + * + * @since 5.1 + */ +public final class RedisUtils { + + private static final String SECTION = "server"; + + private static final String VERSION_PROPERTY = "redis_version"; + + private static final int MAJOR_VERSION_TO_COMPARE = 4; + + private static Boolean unlinkAvailable; + + /** + * Perform an {@code INFO} command on the provided {@link RedisOperations} to check + * the Redis server version to be sure that {@code UNLINK} is available or not. + * @param redisOperations the {@link RedisOperations} to perform {@code INFO} command. + * @return true or false if {@code UNLINK} Redis command is available or not. + * @throws IllegalStateException when {@code INFO} returns null from the Redis. + */ + public static boolean isUnlinkAvailable(RedisOperations redisOperations) { + if (unlinkAvailable == null) { + Properties info = redisOperations.execute( + (RedisCallback) connection -> connection.serverCommands().info(SECTION)); + if (info != null) { + int majorVersion = Integer.parseInt(info.getProperty(VERSION_PROPERTY).split("\\.")[0]); + unlinkAvailable = majorVersion >= MAJOR_VERSION_TO_COMPARE; + } + else { + throw new IllegalStateException("The INFO command cannot be used in pipeline/transaction."); + } + } + return unlinkAvailable; + } + + private RedisUtils() { + } + +}