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 b74be6ca43..608e05e07d 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-2024 the original author or authors. + * Copyright 2002-2025 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. @@ -38,13 +38,12 @@ import org.springframework.util.Assert; * @author Gary Russell * @author Artem Bilan * @author Ngoc Nhan + * @author Youbin Wu * * @since 2.1 */ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupStore implements MessageStore { - private static final String GROUP_ID_MUST_NOT_BE_NULL = "'groupId' must not be null"; - protected static final String MESSAGE_KEY_PREFIX = "MESSAGE_"; protected static final String MESSAGE_GROUP_KEY_PREFIX = "GROUP_OF_MESSAGES_"; @@ -206,7 +205,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS } @Override - public void addMessagesToGroup(Object groupId, Message... messages) { + protected void doAddMessagesToGroup(Object groupId, Message... messages) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); Assert.notNull(messages, "'messages' must not be null"); @@ -240,7 +239,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS } @Override - public void removeMessagesFromGroup(Object groupId, Collection> messages) { + protected void doRemoveMessagesFromGroup(Object groupId, Collection> messages) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); Assert.notNull(messages, "'messages' must not be null"); @@ -283,7 +282,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS } @Override - public boolean removeMessageFromGroupById(Object groupId, UUID messageId) { + protected boolean doRemoveMessageFromGroupById(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); @@ -305,7 +304,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS } @Override - public void completeGroup(Object groupId) { + protected void doCompleteGroup(Object groupId) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); MessageGroupMetadata metadata = getGroupMetadata(groupId); if (metadata != null) { @@ -319,7 +318,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS * Remove the MessageGroup with the provided group ID. */ @Override - public void removeMessageGroup(Object groupId) { + protected void doRemoveMessageGroup(Object groupId) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); Object mgm = doRemove(this.groupPrefix + groupId); if (mgm != null) { @@ -337,7 +336,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS } @Override - public void setGroupCondition(Object groupId, String condition) { + protected void doSetGroupCondition(Object groupId, String condition) { MessageGroupMetadata metadata = getGroupMetadata(groupId); if (metadata != null) { metadata.setCondition(condition); @@ -346,7 +345,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS } @Override - public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { + protected void doSetLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); MessageGroupMetadata metadata = getGroupMetadata(groupId); if (metadata == null) { @@ -359,7 +358,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS } @Override - public Message pollMessageFromGroup(Object groupId) { + protected Message doPollMessageFromGroup(Object groupId) { MessageGroupMetadata groupMetadata = getGroupMetadata(groupId); if (groupMetadata != null) { UUID firstId = groupMetadata.firstId(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java index 8496a51bf1..89c8bdb05f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2023 the original author or authors. + * Copyright 2002-2025 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. @@ -19,16 +19,22 @@ package org.springframework.integration.store; import java.util.Arrays; import java.util.Collection; import java.util.LinkedHashSet; +import java.util.UUID; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.integration.support.locks.DefaultLockRegistry; +import org.springframework.integration.support.locks.LockRegistry; +import org.springframework.integration.util.CheckedCallable; +import org.springframework.integration.util.CheckedRunnable; import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.jmx.export.annotation.ManagedOperation; import org.springframework.jmx.export.annotation.ManagedResource; import org.springframework.messaging.Message; +import org.springframework.util.Assert; /** * @author Dave Syer @@ -36,6 +42,7 @@ import org.springframework.messaging.Message; * @author Gary Russell * @author Artem Bilan * @author Christian Tzolov + * @author Youbin Wu * * @since 2.0 */ @@ -43,6 +50,10 @@ import org.springframework.messaging.Message; public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageGroupStore implements MessageGroupStore, Iterable { + protected static final String INTERRUPTED_WHILE_OBTAINING_LOCK = "Interrupted while obtaining lock"; + + protected static final String GROUP_ID_MUST_NOT_BE_NULL = "'groupId' must not be null"; + protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR final private final Lock lock = new ReentrantLock(); @@ -56,6 +67,8 @@ public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageG private boolean timeoutOnIdle; + private LockRegistry lockRegistry = new DefaultLockRegistry(); + protected AbstractMessageGroupStore() { } @@ -109,6 +122,20 @@ public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageG this.lazyLoadMessageGroups = lazyLoadMessageGroups; } + /** + * Specify the type of the {@link LockRegistry} to ensure atomic operations + * @param lockRegistry lockRegistryType + * @since 6.5 + */ + public final void setLockRegistry(LockRegistry lockRegistry) { + Assert.notNull(lockRegistry, "The LockRegistry cannot be null"); + this.lockRegistry = lockRegistry; + } + + protected LockRegistry getLockRegistry() { + return this.lockRegistry; + } + @Override public void registerMessageGroupExpiryCallback(MessageGroupCallback callback) { if (callback instanceof UniqueExpiryCallback) { @@ -195,12 +222,98 @@ public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageG removeMessagesFromGroup(key, Arrays.asList(messages)); } + @Override + public void removeMessagesFromGroup(Object key, Collection> messages) { + Assert.notNull(key, GROUP_ID_MUST_NOT_BE_NULL); + executeLocked(key, () -> doRemoveMessagesFromGroup(key, messages)); + } + + protected abstract void doRemoveMessagesFromGroup(Object key, Collection> messages); + + @Override + public void addMessagesToGroup(Object groupId, Message... messages) { + Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); + executeLocked(groupId, () -> doAddMessagesToGroup(groupId, messages)); + } + + protected abstract void doAddMessagesToGroup(Object groupId, Message... messages); + @Override public MessageGroup addMessageToGroup(Object groupId, Message message) { addMessagesToGroup(groupId, message); return getMessageGroup(groupId); } + @Override + public void removeMessageGroup(Object groupId) { + Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); + executeLocked(groupId, () -> doRemoveMessageGroup(groupId)); + } + + protected abstract void doRemoveMessageGroup(Object groupId); + + @Override + public boolean removeMessageFromGroupById(Object groupId, UUID messageId) { + Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); + return executeLocked(groupId, () -> doRemoveMessageFromGroupById(groupId, messageId)); + } + + protected boolean doRemoveMessageFromGroupById(Object groupId, UUID messageId) { + throw new UnsupportedOperationException("Not supported for this store"); + } + + @Override + public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { + Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); + executeLocked(groupId, () -> doSetLastReleasedSequenceNumberForGroup(groupId, sequenceNumber)); + } + + protected abstract void doSetLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber); + + @Override + public void completeGroup(Object groupId) { + Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); + executeLocked(groupId, () -> doCompleteGroup(groupId)); + } + + protected abstract void doCompleteGroup(Object groupId); + + @Override + public void setGroupCondition(Object groupId, String condition) { + Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); + executeLocked(groupId, () -> doSetGroupCondition(groupId, condition)); + } + + protected abstract void doSetGroupCondition(Object groupId, String condition); + + @Override + public Message pollMessageFromGroup(Object groupId) { + Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); + return executeLocked(groupId, () -> doPollMessageFromGroup(groupId)); + } + + protected abstract Message doPollMessageFromGroup(Object groupId); + + protected T executeLocked(Object groupId, CheckedCallable runnable) { + try { + return this.lockRegistry.executeLocked(groupId, runnable); + } + catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(INTERRUPTED_WHILE_OBTAINING_LOCK, ex); + } + } + + protected void executeLocked(Object groupId, CheckedRunnable runnable) { + try { + this.lockRegistry.executeLocked(groupId, runnable); + } + catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(INTERRUPTED_WHILE_OBTAINING_LOCK, ex); + } + } + private void expire(MessageGroup group) { RuntimeException exception = null; 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 5f6cd4be70..230b33cb48 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2024 the original author or authors. + * Copyright 2002-2025 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. @@ -45,6 +45,7 @@ import org.springframework.util.CollectionUtils; * @author Gary Russell * @author Ryan Barker * @author Artem Bilan + * @author Youbin Wu * * @since 2.0 */ @@ -55,8 +56,6 @@ public class SimpleMessageStore extends AbstractMessageGroupStore private static final String UPPER_BOUND_MUST_NOT_BE_NULL = "'upperBound' must not be null."; - private static final String INTERRUPTED_WHILE_OBTAINING_LOCK = "Interrupted while obtaining lock"; - private final ConcurrentMap> idToMessage = new ConcurrentHashMap<>(); private final ConcurrentMap groupIdToMessageGroup = new ConcurrentHashMap<>(); @@ -71,12 +70,8 @@ public class SimpleMessageStore extends AbstractMessageGroupStore private final long upperBoundTimeout; - private LockRegistry lockRegistry; - private boolean copyOnGet = false; - private volatile boolean isUsed; - /** * Creates a SimpleMessageStore with a maximum size limited by the given capacity, or unlimited size if the given * capacity is less than 1. The capacities are applied independently to messages stored via @@ -133,8 +128,8 @@ public class SimpleMessageStore extends AbstractMessageGroupStore this.individualUpperBound = new UpperBound(individualCapacity); this.individualCapacity = individualCapacity; this.groupCapacity = groupCapacity; - this.lockRegistry = lockRegistry; this.upperBoundTimeout = upperBoundTimeout; + setLockRegistry(lockRegistry); } /** @@ -162,12 +157,6 @@ public class SimpleMessageStore extends AbstractMessageGroupStore this.copyOnGet = copyOnGet; } - public void setLockRegistry(LockRegistry lockRegistry) { - Assert.notNull(lockRegistry, "The LockRegistry cannot be null"); - Assert.isTrue(!(this.isUsed), "Cannot change the lock registry after the store has been used"); - this.lockRegistry = lockRegistry; - } - @Override public void setLazyLoadMessageGroups(boolean lazyLoadMessageGroups) { throw new UnsupportedOperationException("The lazy-load isn't supported for in-memory 'SimpleMessageStore'"); @@ -181,7 +170,6 @@ public class SimpleMessageStore extends AbstractMessageGroupStore @Override public Message addMessage(Message message) { - this.isUsed = true; if (!this.individualUpperBound.tryAcquire(this.upperBoundTimeout)) { throw new MessagingException(getClass().getSimpleName() + " was out of capacity (" @@ -246,7 +234,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore @Override protected MessageGroup copy(MessageGroup group) { Object groupId = group.getGroupId(); - Lock lock = this.lockRegistry.obtain(groupId); + Lock lock = getLockRegistry().obtain(groupId); try { lock.lockInterruptibly(); try { @@ -261,9 +249,9 @@ public class SimpleMessageStore extends AbstractMessageGroupStore lock.unlock(); } } - catch (InterruptedException e) { + catch (InterruptedException ex) { Thread.currentThread().interrupt(); - throw new MessagingException(INTERRUPTED_WHILE_OBTAINING_LOCK, e); + throw new IllegalStateException(INTERRUPTED_WHILE_OBTAINING_LOCK, ex); } } @@ -272,7 +260,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore Assert.notNull(groupId, "'groupId' must not be null"); Assert.notNull(messages, "'messages' must not be null"); - Lock lock = this.lockRegistry.obtain(groupId); + Lock lock = getLockRegistry().obtain(groupId); try { lock.lockInterruptibly(); boolean unlocked = false; @@ -314,71 +302,50 @@ public class SimpleMessageStore extends AbstractMessageGroupStore } } } - catch (InterruptedException e) { + catch (InterruptedException ex) { Thread.currentThread().interrupt(); - throw new MessagingException(INTERRUPTED_WHILE_OBTAINING_LOCK, e); + throw new IllegalStateException(INTERRUPTED_WHILE_OBTAINING_LOCK, ex); } } - private MessagingException outOfCapacityException(Object groupId) { - return new MessagingException(getClass().getSimpleName() + + @Override + protected void doAddMessagesToGroup(Object groupId, Message... messages) { + // No implementation: the addMessagesToGroup() fully uses locking algorithm. + } + + private IllegalStateException outOfCapacityException(Object groupId) { + return new IllegalStateException(getClass().getSimpleName() + " was out of capacity (" + this.groupCapacity + ") for group '" + groupId + "', try constructing it with a larger number."); } @Override - public void removeMessageGroup(Object groupId) { - Lock lock = this.lockRegistry.obtain(groupId); - try { - lock.lockInterruptibly(); - try { - MessageGroup messageGroup = this.groupIdToMessageGroup.remove(groupId); - if (messageGroup != null) { - UpperBound upperBound = this.groupToUpperBound.remove(groupId); - Assert.state(upperBound != null, UPPER_BOUND_MUST_NOT_BE_NULL); - upperBound.release(this.groupCapacity); - } - } - finally { - lock.unlock(); - } - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new MessagingException(INTERRUPTED_WHILE_OBTAINING_LOCK, e); + protected void doRemoveMessageGroup(Object groupId) { + MessageGroup messageGroup = this.groupIdToMessageGroup.remove(groupId); + if (messageGroup != null) { + UpperBound upperBound = this.groupToUpperBound.remove(groupId); + Assert.state(upperBound != null, UPPER_BOUND_MUST_NOT_BE_NULL); + upperBound.release(this.groupCapacity); } } @Override - public void removeMessagesFromGroup(Object groupId, Collection> messages) { - 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(s) from the MessageGroup"); - UpperBound upperBound = this.groupToUpperBound.get(groupId); - Assert.state(upperBound != null, UPPER_BOUND_MUST_NOT_BE_NULL); - boolean modified = false; - for (Message messageToRemove : messages) { - if (group.remove(messageToRemove)) { - upperBound.release(); - modified = true; - } - } - if (modified) { - group.setLastModified(System.currentTimeMillis()); - } - } - finally { - lock.unlock(); + protected void doRemoveMessagesFromGroup(Object groupId, Collection> messages) { + MessageGroup group = this.groupIdToMessageGroup.get(groupId); + Assert.notNull(group, + () -> MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' " + + "can not be located while attempting to remove Message(s) from the MessageGroup"); + UpperBound upperBound = this.groupToUpperBound.get(groupId); + Assert.state(upperBound != null, UPPER_BOUND_MUST_NOT_BE_NULL); + boolean modified = false; + for (Message messageToRemove : messages) { + if (group.remove(messageToRemove)) { + upperBound.release(); + modified = true; } } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new MessagingException(INTERRUPTED_WHILE_OBTAINING_LOCK, e); + if (modified) { + group.setLastModified(System.currentTimeMillis()); } } @@ -397,35 +364,22 @@ public class SimpleMessageStore extends AbstractMessageGroupStore } @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(); + protected boolean doRemoveMessageFromGroupById(Object groupId, UUID messageId) { + 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; } } - catch (InterruptedException ex) { - Thread.currentThread().interrupt(); - throw new MessagingException(INTERRUPTED_WHILE_OBTAINING_LOCK, ex); - } + return false; } @Override @@ -434,7 +388,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore } @Override - public void setGroupCondition(Object groupId, String condition) { + protected void doSetGroupCondition(Object groupId, String condition) { MessageGroup group = this.groupIdToMessageGroup.get(groupId); if (group != null) { group.setCondition(condition); @@ -442,53 +396,27 @@ public class SimpleMessageStore extends AbstractMessageGroupStore } @Override - public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { - 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 set 'lastReleasedSequenceNumber'"); - group.setLastReleasedMessageSequenceNumber(sequenceNumber); - group.setLastModified(System.currentTimeMillis()); - } - finally { - lock.unlock(); - } - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new MessagingException(INTERRUPTED_WHILE_OBTAINING_LOCK, e); - } + protected void doSetLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { + MessageGroup group = this.groupIdToMessageGroup.get(groupId); + Assert.notNull(group, + () -> MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' " + + "can not be located while attempting to set 'lastReleasedSequenceNumber'"); + group.setLastReleasedMessageSequenceNumber(sequenceNumber); + group.setLastModified(System.currentTimeMillis()); } @Override - public void completeGroup(Object groupId) { - 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 complete the MessageGroup"); - group.complete(); - group.setLastModified(System.currentTimeMillis()); - } - finally { - lock.unlock(); - } - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new MessagingException(INTERRUPTED_WHILE_OBTAINING_LOCK, e); - } + protected void doCompleteGroup(Object groupId) { + MessageGroup group = this.groupIdToMessageGroup.get(groupId); + Assert.notNull(group, + () -> MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' " + + "can not be located while attempting to complete the MessageGroup"); + group.complete(); + group.setLastModified(System.currentTimeMillis()); } @Override - public Message pollMessageFromGroup(Object groupId) { + protected Message doPollMessageFromGroup(Object groupId) { Collection> messageList = getMessageGroup(groupId).getMessages(); Message message = null; if (!CollectionUtils.isEmpty(messageList)) { @@ -521,7 +449,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore } public void clearMessageGroup(Object groupId) { - Lock lock = this.lockRegistry.obtain(groupId); + Lock lock = getLockRegistry().obtain(groupId); try { lock.lockInterruptibly(); try { @@ -539,9 +467,9 @@ public class SimpleMessageStore extends AbstractMessageGroupStore lock.unlock(); } } - catch (InterruptedException e) { + catch (InterruptedException ex) { Thread.currentThread().interrupt(); - throw new MessagingException(INTERRUPTED_WHILE_OBTAINING_LOCK, e); + throw new IllegalStateException(INTERRUPTED_WHILE_OBTAINING_LOCK, ex); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java index 757ed23a32..1c472142f1 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2022 the original author or authors. + * Copyright 2002-2025 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. @@ -56,6 +56,7 @@ import static org.mockito.Mockito.verify; * @author Gary Russell * @author Artem Bilan * @author Meherzad Lahewala + * @author Youbin Wu * * @since 2.2 * @@ -350,7 +351,7 @@ public class AbstractCorrelatingMessageHandlerTests { SimpleMessageStore messageStore = new SimpleMessageStore() { @Override - public void removeMessageGroup(Object groupId) { + protected void doRemoveMessageGroup(Object groupId) { throw new RuntimeException("intentional"); } }; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java index 988de55451..d12df63758 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2024 the original author or authors. + * Copyright 2002-2025 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. @@ -22,9 +22,8 @@ import java.util.Collections; import java.util.Iterator; import java.util.List; -import org.junit.Test; +import org.junit.jupiter.api.Test; -import org.springframework.integration.store.MessageGroupStore.MessageGroupCallback; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.util.ReflectionTestUtils; @@ -35,22 +34,22 @@ import static org.assertj.core.api.Assertions.assertThat; * @author Dave Syer * @author Gary Russell * @author Artem Bilan + * @author Youbin Wu */ public class MessageStoreTests { @Test - public void shouldRegisterCallbacks() throws Exception { + public void shouldRegisterCallbacks() { TestMessageStore store = new TestMessageStore(); - store.setExpiryCallbacks(Collections.singletonList((messageGroupStore, group) -> { + store.setExpiryCallbacks(Collections.singletonList((messageGroupStore, group) -> { })); assertThat(((Collection) ReflectionTestUtils.getField(store, "expiryCallbacks")).size()).isEqualTo(1); } @Test - public void shouldExpireMessageGroup() throws Exception { - + public void shouldExpireMessageGroup() { TestMessageStore store = new TestMessageStore(); - final List list = new ArrayList(); + final List list = new ArrayList<>(); store.registerMessageGroupExpiryCallback((messageGroupStore, group) -> { list.add(group.getOne().getPayload().toString()); messageGroupStore.removeMessageGroup(group.getGroupId()); @@ -63,13 +62,13 @@ public class MessageStoreTests { } @Test - public void testGroupCount() throws Exception { + public void testGroupCount() { TestMessageStore store = new TestMessageStore(); assertThat(store.getMessageGroupCount()).isEqualTo(1); } @Test - public void testGroupSizes() throws Exception { + public void testGroupSizes() { TestMessageStore store = new TestMessageStore(); assertThat(store.getMessageCountForAllMessageGroups()).isEqualTo(1); } @@ -91,7 +90,7 @@ public class MessageStoreTests { } @Override - public void addMessagesToGroup(Object groupId, Message... messages) { + protected void doAddMessagesToGroup(Object groupId, Message... messages) { throw new UnsupportedOperationException(); } @@ -101,30 +100,30 @@ public class MessageStoreTests { } @Override - public void removeMessagesFromGroup(Object key, Collection> messages) { + protected void doRemoveMessagesFromGroup(Object key, Collection> messages) { throw new UnsupportedOperationException(); } @Override - public void removeMessageGroup(Object correlationKey) { + protected void doRemoveMessageGroup(Object correlationKey) { if (correlationKey.equals(testMessages.getGroupId())) { removed = true; } } @Override - public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { + protected void doSetLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { throw new UnsupportedOperationException(); } @Override - public void completeGroup(Object groupId) { + protected void doCompleteGroup(Object groupId) { throw new UnsupportedOperationException(); } @Override - public Message pollMessageFromGroup(Object groupId) { + protected Message doPollMessageFromGroup(Object groupId) { return null; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java index 7e87f040ac..bb196a0af7 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2024 the original author or authors. + * Copyright 2002-2025 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. @@ -34,6 +34,7 @@ import org.springframework.test.util.ReflectionTestUtils; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatExceptionOfType; +import static org.assertj.core.api.Assertions.assertThatIllegalStateException; /** * @author Iwein Fuld @@ -45,7 +46,6 @@ import static org.assertj.core.api.Assertions.assertThatExceptionOfType; public class SimpleMessageStoreTests { @Test - @SuppressWarnings("unchecked") public void shouldRetainMessage() { SimpleMessageStore store = new SimpleMessageStore(); Message testMessage1 = MessageBuilder.withPayload("foo").build(); @@ -124,7 +124,7 @@ public class SimpleMessageStoreTests { Message testMessage1 = MessageBuilder.withPayload("foo").build(); Message testMessage2 = MessageBuilder.withPayload("bar").build(); store.addMessageToGroup("foo", testMessage1); - assertThatExceptionOfType(MessagingException.class) + assertThatIllegalStateException() .isThrownBy(() -> store.addMessageToGroup("foo", testMessage2)); } @@ -158,7 +158,7 @@ public class SimpleMessageStoreTests { SimpleMessageStore store2 = new SimpleMessageStore(1, 1, 1); store2.addMessageToGroup("foo", MessageBuilder.withPayload("foo").build()); - assertThatExceptionOfType(MessagingException.class) + assertThatIllegalStateException() .isThrownBy(() -> store2.addMessageToGroup("foo", MessageBuilder.withPayload("bar").build())); } @@ -178,13 +178,13 @@ public class SimpleMessageStoreTests { Message testMessage2 = MessageBuilder.withPayload("bar").build(); store.addMessageToGroup("foo", testMessage1); - assertThatExceptionOfType(MessagingException.class) + assertThatIllegalStateException() .isThrownBy(() -> store.addMessageToGroup("foo", testMessage2)) .withMessageContaining("was out of capacity (1) for group 'foo'"); store.removeMessagesFromGroup("foo", testMessage2); - assertThatExceptionOfType(MessagingException.class) + assertThatIllegalStateException() .isThrownBy(() -> store.addMessageToGroup("foo", testMessage2)) .withMessageContaining("was out of capacity (1) for group 'foo'"); 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 db47f54235..73d0f99760 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2024 the original author or authors. + * Copyright 2002-2025 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. @@ -80,6 +80,7 @@ import org.springframework.util.StringUtils; * @author Gary Russell * @author Artem Bilan * @author Ngoc Nhan + * @author Youbin Wu * * @since 2.0 */ @@ -472,7 +473,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore } @Override - public void addMessagesToGroup(Object groupId, Message... messages) { + protected void doAddMessagesToGroup(Object groupId, Message... messages) { String groupKey = getKey(groupId); MessageGroupMetadata groupMetadata = getGroupMetadata(groupKey); @@ -576,7 +577,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore } @Override - public void removeMessagesFromGroup(Object groupId, Collection> messages) { + protected void doRemoveMessagesFromGroup(Object groupId, Collection> messages) { Assert.notNull(groupId, "'groupId' must not be null"); Assert.notNull(messages, "'messages' must not be null"); @@ -621,7 +622,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore } @Override - public boolean removeMessageFromGroupById(Object groupId, UUID messageId) { + protected boolean doRemoveMessageFromGroupById(Object groupId, UUID messageId) { String groupKey = getKey(groupId); String messageKey = getKey(messageId); int messageToGroupRemoved = @@ -634,7 +635,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore } @Override - public void removeMessageGroup(Object groupId) { + protected void doRemoveMessageGroup(Object groupId) { String groupKey = getKey(groupId); this.jdbcTemplate.update(getQuery(Query.DELETE_MESSAGES_FROM_GROUP), @@ -653,7 +654,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore } @Override - public void completeGroup(Object groupId) { + protected void doCompleteGroup(Object groupId) { final String groupKey = getKey(groupId); if (logger.isDebugEnabled()) { @@ -664,7 +665,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore } @Override - public void setGroupCondition(Object groupId, String condition) { + protected void doSetGroupCondition(Object groupId, String condition) { Assert.notNull(groupId, "'groupId' must not be null"); String groupKey = getKey(groupId); Timestamp updatedDate = new Timestamp(System.currentTimeMillis()); @@ -675,7 +676,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore } @Override - public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { + protected void doSetLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { Assert.notNull(groupId, "'groupId' must not be null"); String groupKey = getKey(groupId); @@ -687,7 +688,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore } @Override - public Message pollMessageFromGroup(Object groupId) { + protected Message doPollMessageFromGroup(Object groupId) { String key = getKey(groupId); Message polledMessage = doPollForMessage(key); diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/AbstractConfigurableMongoDbMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/AbstractConfigurableMongoDbMessageStore.java index d7cbd4259f..3e23749141 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/AbstractConfigurableMongoDbMessageStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/AbstractConfigurableMongoDbMessageStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2024 the original author or authors. + * Copyright 2014-2025 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. @@ -61,6 +61,7 @@ import org.springframework.util.Assert; * * @author Artem Bilan * @author Adama Sorho + * @author Youbin Wu * * @since 4.0 */ @@ -201,7 +202,7 @@ public abstract class AbstractConfigurableMongoDbMessageStore extends AbstractMe } @Override - public void removeMessageGroup(Object groupId) { + protected void doRemoveMessageGroup(Object groupId) { this.mongoTemplate.remove(groupIdQuery(groupId), this.collectionName); } @@ -250,17 +251,17 @@ public abstract class AbstractConfigurableMongoDbMessageStore extends AbstractMe } @Override - public void removeMessagesFromGroup(Object key, Collection> messages) { + protected void doRemoveMessagesFromGroup(Object key, Collection> messages) { throw NOT_IMPLEMENTED; } @Override - public void setGroupCondition(Object groupId, String condition) { + protected void doSetGroupCondition(Object groupId, String condition) { throw NOT_IMPLEMENTED; } @Override - public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { + protected void doSetLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { throw NOT_IMPLEMENTED; } @@ -270,7 +271,7 @@ public abstract class AbstractConfigurableMongoDbMessageStore extends AbstractMe } @Override - public void completeGroup(Object groupId) { + protected void doCompleteGroup(Object groupId) { throw NOT_IMPLEMENTED; } @@ -280,7 +281,7 @@ public abstract class AbstractConfigurableMongoDbMessageStore extends AbstractMe } @Override - public void addMessagesToGroup(Object groupId, Message... messages) { + protected void doAddMessagesToGroup(Object groupId, Message... messages) { throw NOT_IMPLEMENTED; } 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 6176a2ccaf..04e9d860c8 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 @@ -1,5 +1,5 @@ /* - * Copyright 2013-2024 the original author or authors. + * Copyright 2013-2025 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. @@ -54,6 +54,7 @@ import org.springframework.util.Assert; * @author Artem Bilan * @author Gary Russell * @author Ngoc Nhan + * @author Youbin Wu * * @since 3.0 */ @@ -147,7 +148,7 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb } @Override - public void addMessagesToGroup(Object groupId, Message... messages) { + protected void doAddMessagesToGroup(Object groupId, Message... messages) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); Assert.notNull(messages, "'message' must not be null"); @@ -183,7 +184,7 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb } @Override - public void removeMessagesFromGroup(Object groupId, Collection> messages) { + protected void doRemoveMessagesFromGroup(Object groupId, Collection> messages) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); Assert.notNull(messages, "'messageToRemove' must not be null"); @@ -215,7 +216,7 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb } @Override - public boolean removeMessageFromGroupById(Object groupId, UUID messageId) { + protected boolean doRemoveMessageFromGroupById(Object groupId, UUID messageId) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); Assert.notNull(messageId, "'messageId' must not be null"); Query query = @@ -234,7 +235,7 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb } @Override - public Message pollMessageFromGroup(final Object groupId) { + protected Message doPollMessageFromGroup(final Object groupId) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); Sort sort = Sort.by(MessageDocumentFields.LAST_MODIFIED_TIME, MessageDocumentFields.SEQUENCE); @@ -249,17 +250,17 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb } @Override - public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { + protected void doSetLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { updateGroup(groupId, lastModifiedUpdate().set(MessageDocumentFields.LAST_RELEASED_SEQUENCE, sequenceNumber)); } @Override - public void setGroupCondition(Object groupId, String condition) { + protected void doSetGroupCondition(Object groupId, String condition) { updateGroup(groupId, lastModifiedUpdate().set("condition", condition)); } @Override - public void completeGroup(Object groupId) { + protected void doCompleteGroup(Object groupId) { updateGroup(groupId, lastModifiedUpdate().set(MessageDocumentFields.COMPLETE, true)); } diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbChannelMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbChannelMessageStore.java index 76ee7da77a..b9c0ed38b5 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbChannelMessageStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbChannelMessageStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2023 the original author or authors. + * Copyright 2014-2025 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. @@ -44,6 +44,7 @@ import org.springframework.util.Assert; * * @author Artem Bilan * @author Adama Sorho + * @author Youbin Wu * * @since 4.0 */ @@ -132,7 +133,7 @@ public class MongoDbChannelMessageStore extends AbstractConfigurableMongoDbMessa } @Override - public Message pollMessageFromGroup(Object groupId) { + protected Message doPollMessageFromGroup(Object groupId) { Assert.notNull(groupId, "'groupId' must not be null"); Sort sort = Sort.by(MessageDocumentFields.LAST_MODIFIED_TIME, MessageDocumentFields.SEQUENCE); 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 617d7e4c11..62baaf27bb 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2024 the original author or authors. + * Copyright 2002-2025 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. @@ -97,6 +97,7 @@ import org.springframework.util.StringUtils; * @author Jodie StJohn * @author Gary Russell * @author Artem Bilan + * @author Youbin Wu * * @since 2.1 */ @@ -294,7 +295,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore } @Override - public void addMessagesToGroup(Object groupId, Message... messages) { + protected void doAddMessagesToGroup(Object groupId, Message... messages) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); Assert.notNull(messages, "'message' must not be null"); Query query = whereGroupIdOrder(groupId); @@ -328,7 +329,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore } @Override - public void removeMessagesFromGroup(Object groupId, Collection> messages) { + protected void doRemoveMessagesFromGroup(Object groupId, Collection> messages) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); Assert.notNull(messages, "'messageToRemove' must not be null"); @@ -367,7 +368,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore } @Override - public boolean removeMessageFromGroupById(Object groupId, UUID messageId) { + protected boolean doRemoveMessageFromGroupById(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) @@ -375,7 +376,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore } @Override - public void removeMessageGroup(Object groupId) { + protected void doRemoveMessageGroup(Object groupId) { this.template.remove(whereGroupIdIs(groupId), this.collectionName); } @@ -395,7 +396,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore } @Override - public Message pollMessageFromGroup(final Object groupId) { + protected Message doPollMessageFromGroup(final Object groupId) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); Query query = whereGroupIdIs(groupId).with(Sort.by(GROUP_UPDATE_TIMESTAMP_KEY, SEQUENCE)); MessageWrapper messageWrapper = this.template.findAndRemove(query, MessageWrapper.class, this.collectionName); @@ -415,17 +416,17 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore } @Override - public void setGroupCondition(Object groupId, String condition) { + protected void doSetGroupCondition(Object groupId, String condition) { updateGroup(groupId, lastModifiedUpdate().set("_condition", condition)); } @Override - public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { + protected void doSetLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { updateGroup(groupId, lastModifiedUpdate().set(LAST_RELEASED_SEQUENCE_NUMBER, sequenceNumber)); } @Override - public void completeGroup(Object groupId) { + protected void doCompleteGroup(Object groupId) { this.updateGroup(groupId, lastModifiedUpdate().set(GROUP_COMPLETE_KEY, true)); } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java index 68bd2d112c..d54cf724c0 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2007-2024 the original author or authors. + * Copyright 2007-2025 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. @@ -493,6 +493,15 @@ class RedisMessageGroupStoreTests implements RedisContainerTest { assertThat(store.messageGroupSize("2")).isEqualTo(1); } + @Test + public void testMessageGroupCondition() { + String groupId = "X"; + Message message = MessageBuilder.withPayload("foo").build(); + store.addMessagesToGroup(groupId, message); + store.setGroupCondition(groupId, "testCondition"); + assertThat(store.getMessageGroup(groupId).getCondition()).isEqualTo("testCondition"); + } + private record Foo(String foo) { } diff --git a/src/reference/antora/modules/ROOT/pages/message-store.adoc b/src/reference/antora/modules/ROOT/pages/message-store.adoc index 3688e452bd..8dfbfab10e 100644 --- a/src/reference/antora/modules/ROOT/pages/message-store.adoc +++ b/src/reference/antora/modules/ROOT/pages/message-store.adoc @@ -158,3 +158,13 @@ It also allows the end marker to arrive at the aggregator before all the other r In addition, for configuration convenience, a `GroupConditionProvider` contract has been introduced. The `AbstractCorrelatingMessageHandler` checks if the provided `ReleaseStrategy` implements this interface and extracts a `conditionSupplier` for group condition evaluation logic. + +[[use-lock-registry]] +== Use `LockRegistry` + +Starting with version 6.5, the `AbstractMessageGroupStore` abstraction operates a metadata of message group with a lock. +This lock acquires the groupId and generated by `LockRegister`. +Its purpose is to operate on the atomicity of messages and message groups. +In multiple threads, adding or removing messages or updating metadata at the same time, some implementations may have message group errors if the lock is missing. +By default, the `DefaultLockRegistry` is used, any `LockRegister` can be injected via `AbstractMessageGroupStore.setLockRegistry()`, usually an implementation for the same persistent store. +See more xref:distributed-locks.adoc[Distributed Locks] for more information. \ No newline at end of file diff --git a/src/reference/antora/modules/ROOT/pages/whats-new.adoc b/src/reference/antora/modules/ROOT/pages/whats-new.adoc index 2726f6db8f..5216a7a6ab 100644 --- a/src/reference/antora/modules/ROOT/pages/whats-new.adoc +++ b/src/reference/antora/modules/ROOT/pages/whats-new.adoc @@ -25,4 +25,10 @@ See xref:control-bus.adoc[Control Bus] for more information. The `AbstractCorrelatingMessageHandler` does not throw an `IllegalArgumentException` for the collection of payloads as a result of the `MessageGroupProcessor`. Instead, such a collection is wrapped into a single reply message. -See xref:aggregator.adoc[Aggregator] for more information. \ No newline at end of file +See xref:aggregator.adoc[Aggregator] for more information. + +[[x6.4-message-store-with-locks]] +== The `LockRegistry` in the `MessageStore` + +The `AbstractMessageGroupStore` now can be configured with a `LockRegistry` to perform series of persistent operation atomically. +See xref:message-store.adoc#use-lock-registry[Use LockRegistry] for more information. \ No newline at end of file