diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java index 8d515f270f..0c57dd7d05 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java @@ -19,7 +19,6 @@ package org.springframework.integration.handler; import java.io.Serial; import java.io.Serializable; import java.time.Instant; -import java.util.Collection; import java.util.Collections; import java.util.Date; import java.util.List; @@ -43,7 +42,6 @@ import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.expression.ExpressionUtils; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; -import org.springframework.integration.store.MessageStore; import org.springframework.integration.store.SimpleMessageStore; import org.springframework.integration.support.management.IntegrationManagedResource; import org.springframework.jmx.export.annotation.ManagedResource; @@ -300,9 +298,7 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement if (this.messageStore == null) { this.messageStore = new SimpleMessageStore(); } - else { - Assert.isInstanceOf(MessageStore.class, this.messageStore); - } + this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory()); this.releaseHandler = createReleaseMessageTask(); } @@ -464,7 +460,7 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement } private Message getMessageById(UUID messageId) { - Message theMessage = ((MessageStore) this.messageStore).getMessage(messageId); + Message theMessage = this.messageStore.getMessageFromGroup(this.messageGroupId, messageId); if (theMessage == null) { logger.debug(() -> "No message in the Message Store for id: " + messageId + @@ -537,11 +533,9 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement } private void doReleaseMessage(Message message) { - if (removeDelayedMessageFromMessageStore(message) + if (this.messageStore.removeMessageFromGroupById(this.messageGroupId, message.getHeaders().getId()) || this.deliveries.get(ObjectUtils.getIdentityHexString(message)).get() > 0) { - if (!(this.messageStore instanceof SimpleMessageStore)) { - this.messageStore.removeMessagesFromGroup(this.messageGroupId, message); - } + handleMessageInternal(message); } else { @@ -550,24 +544,6 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement } } - private boolean removeDelayedMessageFromMessageStore(Message message) { - if (this.messageStore instanceof SimpleMessageStore) { - synchronized (this.messageGroupId) { - Collection> messages = this.messageStore.getMessageGroup(this.messageGroupId).getMessages(); - if (messages.contains(message)) { - this.messageStore.removeMessagesFromGroup(this.messageGroupId, message); - return true; - } - else { - return false; - } - } - } - else { - return ((MessageStore) this.messageStore).removeMessage(message.getHeaders().getId()) != null; - } - } - @Override public int getDelayedMessageCount() { return this.messageStore.messageGroupSize(this.messageGroupId); 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 68505a830b..eea05f4e77 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-2021 the original author or authors. + * Copyright 2002-2023 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. @@ -27,6 +27,7 @@ import java.util.stream.Collectors; import java.util.stream.Stream; import org.springframework.jmx.export.annotation.ManagedAttribute; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.util.Assert; @@ -263,6 +264,42 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS } } + @Override + @Nullable + public Message getMessageFromGroup(Object groupId, UUID messageId) { + Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); + Assert.notNull(messageId, "'messageId' must not be null"); + Object object = doRetrieve(this.messagePrefix + groupId + '_' + messageId); + if (object != null) { + return extractMessage(object); + } + else { + return null; + } + } + + @Override + public boolean removeMessageFromGroupById(Object groupId, UUID messageId) { + Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); + Assert.notNull(messageId, "'messageId' must not be null"); + Object mgm = doRetrieve(this.groupPrefix + groupId); + if (mgm != null) { + Assert.isInstanceOf(MessageGroupMetadata.class, mgm); + MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm; + + if (messageGroupMetadata.getMessageIds().contains(messageId)) { + messageGroupMetadata.remove(messageId); + String groupToMessageId = this.messagePrefix + groupId + '_' + messageId; + if (doRemove(groupToMessageId) != null) { + messageGroupMetadata.setLastModified(System.currentTimeMillis()); + doStore(this.groupPrefix + groupId, messageGroupMetadata); + return true; + } + } + } + return false; + } + @Override public void completeGroup(Object groupId) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java index 48c90ed6c3..21a127428a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2021 the original author or authors. + * Copyright 2002-2023 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. @@ -18,10 +18,12 @@ package org.springframework.integration.store; import java.util.Collection; import java.util.Iterator; +import java.util.UUID; import java.util.stream.Stream; import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.jmx.export.annotation.ManagedOperation; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; /** @@ -71,6 +73,30 @@ public interface MessageGroupStore extends BasicMessageGroupStore { */ void removeMessagesFromGroup(Object key, Message... messages); + /** + * Retrieve a {@link Message} from a group by id. + * Return {@code null} if message does not belong to the requested group. + * @param groupId The groupId for the group containing the message. + * @param messageId The message id. + * @return message by id if it belongs to requested group. + * @since 6.1.5 + */ + @Nullable + default Message getMessageFromGroup(Object groupId, UUID messageId) { + throw new UnsupportedOperationException("Not supported for this store"); + } + + /** + * Deletion the message from the group. + * @param groupId The groupId for the group containing the message. + * @param messageId The message id to be removed. + * @return true if message has been removed. + * @since 6.1.5 + */ + default boolean removeMessageFromGroupById(Object groupId, UUID messageId) { + throw new UnsupportedOperationException("Not supported for this store"); + } + /** * Register a callback for when a message group is expired through {@link #expireMessageGroups(long)}. * @param callback A callback to execute when a message group is cleaned up. @@ -114,7 +140,7 @@ public interface MessageGroupStore extends BasicMessageGroupStore { /** * Completes this MessageGroup. Completion of the MessageGroup generally means - * that this group should not be allowing any more mutating operation to be performed on it. + * that this group should not be allowing anymore mutating operation to be performed on it. * For example any attempt to add/remove new Message form the group should not be allowed. * @param groupId The group identifier. */ diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java index 6ae4e33170..fd53e3bb79 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java @@ -28,6 +28,7 @@ import org.springframework.integration.support.locks.DefaultLockRegistry; import org.springframework.integration.support.locks.LockRegistry; import org.springframework.integration.util.UpperBound; import org.springframework.jmx.export.annotation.ManagedAttribute; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessagingException; import org.springframework.util.Assert; @@ -382,6 +383,52 @@ public class SimpleMessageStore extends AbstractMessageGroupStore } } + @Override + @Nullable + public Message getMessageFromGroup(Object groupId, UUID messageId) { + MessageGroup group = this.groupIdToMessageGroup.get(groupId); + Assert.notNull(group, + () -> MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' does not exists"); + for (Message message : group.getMessages()) { + if (messageId.equals(message.getHeaders().getId())) { + return message; + } + } + return null; + } + + @Override + public boolean removeMessageFromGroupById(Object groupId, UUID messageId) { + Lock lock = this.lockRegistry.obtain(groupId); + try { + lock.lockInterruptibly(); + try { + MessageGroup group = this.groupIdToMessageGroup.get(groupId); + Assert.notNull(group, + () -> MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' " + + "can not be located while attempting to remove Message from the MessageGroup"); + UpperBound upperBound = this.groupToUpperBound.get(groupId); + Assert.state(upperBound != null, UPPER_BOUND_MUST_NOT_BE_NULL); + for (Message message : group.getMessages()) { + if (messageId.equals(message.getHeaders().getId())) { + group.remove(message); + upperBound.release(); + group.setLastModified(System.currentTimeMillis()); + return true; + } + } + return false; + } + finally { + lock.unlock(); + } + } + catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + throw new MessagingException(INTERRUPTED_WHILE_OBTAINING_LOCK, ex); + } + } + @Override public Iterator iterator() { return new HashSet<>(this.groupIdToMessageGroup.values()).iterator(); diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcMessageStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcMessageStore.java index 2288e7d8a3..aa41c693bf 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcMessageStore.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcMessageStore.java @@ -52,6 +52,7 @@ import org.springframework.jdbc.core.SingleColumnRowMapper; import org.springframework.jdbc.support.lob.DefaultLobHandler; import org.springframework.jdbc.support.lob.LobHandler; import org.springframework.jmx.export.annotation.ManagedAttribute; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -155,6 +156,14 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa where MESSAGE_ID=? and REGION=? """), + GET_MESSAGE_FROM_GROUP(""" + SELECT m.MESSAGE_ID, m.CREATED_DATE, m.MESSAGE_BYTES + from %PREFIX%MESSAGE m + inner join %PREFIX%GROUP_TO_MESSAGE gm + on m.MESSAGE_ID = gm.MESSAGE_ID + where gm.MESSAGE_ID=? and gm.GROUP_KEY = ? and gm.REGION=? + """), + GET_MESSAGE_COUNT(""" SELECT COUNT(MESSAGE_ID) from %PREFIX%MESSAGE @@ -554,6 +563,31 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa updateMessageGroup(groupKey); } + @Override + @Nullable + public Message getMessageFromGroup(Object groupId, UUID messageId) { + List> list = + this.jdbcTemplate.query(getQuery(Query.GET_MESSAGE_FROM_GROUP), this.mapper, + getKey(messageId), getKey(groupId), this.region); + if (list.isEmpty()) { + return null; + } + return list.get(0); + } + + @Override + public boolean removeMessageFromGroupById(Object groupId, UUID messageId) { + String groupKey = getKey(groupId); + String messageKey = getKey(messageId); + int messageToGroupRemoved = + this.jdbcTemplate.update(getQuery(Query.REMOVE_MESSAGE_FROM_GROUP), groupKey, messageKey, this.region); + if (messageToGroupRemoved > 0) { + return this.jdbcTemplate.update(getQuery(Query.DELETE_MESSAGE), + messageKey, this.region, messageKey, this.region) > 0; + } + return false; + } + @Override public void removeMessageGroup(Object groupId) { String groupKey = getKey(groupId); diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/DelayerHandlerRescheduleIntegrationTests-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/DelayerHandlerRescheduleIntegrationTests-context.xml index 8f2a9b1ca8..da59932ea1 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/DelayerHandlerRescheduleIntegrationTests-context.xml +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/DelayerHandlerRescheduleIntegrationTests-context.xml @@ -23,7 +23,7 @@ diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/DelayerHandlerRescheduleIntegrationTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/DelayerHandlerRescheduleIntegrationTests.java index 0a9538ba6f..dc3732ff4e 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/DelayerHandlerRescheduleIntegrationTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/DelayerHandlerRescheduleIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2022 the original author or authors. + * Copyright 2002-2023 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. @@ -31,7 +31,6 @@ import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.support.MessageBuilder; -import org.springframework.integration.test.condition.LongRunningTest; import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.util.UUIDConverter; import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase; @@ -47,13 +46,12 @@ import org.springframework.transaction.support.TransactionSynchronization; import org.springframework.transaction.support.TransactionSynchronizationManager; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.fail; +import static org.assertj.core.api.Assertions.assertThatIllegalStateException; /** * @author Artem Bilan * @author Gary Russell */ -@LongRunningTest public class DelayerHandlerRescheduleIntegrationTests { public static final String DELAYER_ID = "delayerWithJdbcMS"; @@ -98,15 +96,9 @@ public class DelayerHandlerRescheduleIntegrationTests { taskScheduler.getScheduledExecutor().awaitTermination(10, TimeUnit.SECONDS); context.close(); - try { - context.getBean("input", MessageChannel.class); - fail("IllegalStateException expected"); - } - catch (Exception e) { - assertThat(e instanceof IllegalStateException).isTrue(); - assertThat(e.getMessage().contains("BeanFactory not initialized or already closed - call 'refresh'")) - .isTrue(); - } + assertThatIllegalStateException() + .isThrownBy(() -> context.getBean("input", MessageChannel.class)) + .withMessageContaining("BeanFactory not initialized or already closed - call 'refresh'"); String delayerMessageGroupId = UUIDConverter.getUUID(DELAYER_ID + ".messageGroupId").toString(); diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java index a4ad345b8e..e88e9c6aac 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java @@ -38,6 +38,7 @@ import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageStore; import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.jmx.export.annotation.ManagedAttribute; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.util.Assert; @@ -200,6 +201,32 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb updateGroup(groupId, lastModifiedUpdate()); } + @Override + @Nullable + public Message getMessageFromGroup(Object groupId, UUID messageId) { + Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); + Assert.notNull(messageId, "'messageId' must not be null"); + Query query = + Query.query( + Criteria.where(MessageDocumentFields.MESSAGE_ID).is(messageId) + .and(MessageDocumentFields.GROUP_ID).is(groupId)); + MessageDocument document = getMongoTemplate().findOne(query, MessageDocument.class, this.collectionName); + return document != null ? document.getMessage() : null; + } + + @Override + public boolean removeMessageFromGroupById(Object groupId, UUID messageId) { + Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); + Assert.notNull(messageId, "'messageId' must not be null"); + Query query = + Query.query( + Criteria.where(MessageDocumentFields.MESSAGE_ID).is(messageId) + .and(MessageDocumentFields.GROUP_ID).is(groupId)); + return getMongoTemplate() + .remove(query, this.collectionName) + .wasAcknowledged(); + } + private void removeMessages(Object groupId, Collection ids) { Query query = groupIdQuery(groupId) .addCriteria(Criteria.where(MessageDocumentFields.MESSAGE_ID).in(ids.toArray())); diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java index b00d75a3da..0b24298b51 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java @@ -356,6 +356,25 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore bulkOperations.execute(); } + @Override + @Nullable + public Message getMessageFromGroup(Object groupId, UUID messageId) { + Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); + Assert.notNull(messageId, "'messageId' must not be null"); + MessageWrapper messageWrapper = + this.template.findOne(whereMessageIdIsAndGroupIdIs(messageId, groupId), + MessageWrapper.class, this.collectionName); + return (messageWrapper != null) ? messageWrapper.getMessage() : null; + } + + @Override + public boolean removeMessageFromGroupById(Object groupId, UUID messageId) { + Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); + Assert.notNull(messageId, "'messageId' must not be null"); + return this.template.remove(whereMessageIdIsAndGroupIdIs(messageId, groupId), this.collectionName) + .wasAcknowledged(); + } + @Override public void removeMessageGroup(Object groupId) { this.template.remove(whereGroupIdIs(groupId), this.collectionName); diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationConfigurableTests-context.xml b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationConfigurableTests-context.xml index 59f4ef33b3..87d5e1e4ce 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationConfigurableTests-context.xml +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationConfigurableTests-context.xml @@ -22,7 +22,7 @@ diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationTests-context.xml b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationTests-context.xml index df1e2b3c9a..1b288ac5ca 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationTests-context.xml +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationTests-context.xml @@ -22,7 +22,7 @@ diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationTests.java index b268fb85c9..5bbf33f309 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2022 the original author or authors. + * Copyright 2013-2023 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. @@ -29,7 +29,6 @@ import org.springframework.integration.mongodb.MongoDbContainerTest; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.support.MessageBuilder; -import org.springframework.integration.test.condition.LongRunningTest; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; @@ -43,7 +42,6 @@ import static org.assertj.core.api.Assertions.assertThat; * * @since 3.0 */ -@LongRunningTest class DelayerHandlerRescheduleIntegrationTests implements MongoDbContainerTest { public static final String DELAYER_ID = "delayerWithMongoMS"; diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/DelayerHandlerRescheduleIntegrationTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/DelayerHandlerRescheduleIntegrationTests-context.xml index dcb5c3068a..247c6b9900 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/DelayerHandlerRescheduleIntegrationTests-context.xml +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/DelayerHandlerRescheduleIntegrationTests-context.xml @@ -7,7 +7,7 @@ + value="#{T (org.springframework.integration.redis.RedisContainerTest).connectionFactory() }"/> @@ -17,7 +17,7 @@ diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/DelayerHandlerRescheduleIntegrationTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/DelayerHandlerRescheduleIntegrationTests.java index 56ebb6ce75..6cfdccbe0e 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/DelayerHandlerRescheduleIntegrationTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/DelayerHandlerRescheduleIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2022 the original author or authors. + * Copyright 2013-2023 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. @@ -29,14 +29,12 @@ import org.springframework.integration.redis.RedisContainerTest; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.support.MessageBuilder; -import org.springframework.integration.test.condition.LongRunningTest; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.fail; /** * @author Artem Bilan @@ -45,8 +43,8 @@ import static org.assertj.core.api.Assertions.fail; * * @since 3.0 */ -@LongRunningTest class DelayerHandlerRescheduleIntegrationTests implements RedisContainerTest { + public static final String DELAYER_ID = "delayerWithRedisMS" + UUID.randomUUID(); @Test @@ -73,15 +71,6 @@ class DelayerHandlerRescheduleIntegrationTests implements RedisContainerTest { assertThat(taskScheduler.getScheduledExecutor().awaitTermination(10, TimeUnit.SECONDS)).isTrue(); context.close(); - try { - context.getBean("input", MessageChannel.class); - fail("IllegalStateException expected"); - } - catch (Exception e) { - assertThat(e).isInstanceOf(IllegalStateException.class); - assertThat(e.getMessage()).contains("BeanFactory not initialized or already closed - call 'refresh'"); - } - assertThat(messageStore.getMessageGroupCount()).isEqualTo(1); assertThat(messageStore.iterator().next().getGroupId()).isEqualTo(delayerMessageGroupId); assertThat(messageStore.messageGroupSize(delayerMessageGroupId)).isEqualTo(2);