Class AbstractMessageGroupStore

java.lang.Object
org.springframework.integration.store.AbstractBatchingMessageGroupStore
org.springframework.integration.store.AbstractMessageGroupStore
All Implemented Interfaces:
Iterable<MessageGroup>, BasicMessageGroupStore, MessageGroupStore
Direct Known Subclasses:
AbstractConfigurableMongoDbMessageStore, AbstractKeyValueMessageStore, JdbcMessageStore, MongoDbMessageStore, SimpleMessageStore
@ManagedResource public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageGroupStore implements MessageGroupStore, Iterable<MessageGroup>
Since:
2.0
Author:
Dave Syer, Oleg Zhurakousky, Gary Russell, Artem Bilan, Christian Tzolov, Youbin Wu

  • Field Details

    • INTERRUPTED_WHILE_OBTAINING_LOCK

      protected static final String INTERRUPTED_WHILE_OBTAINING_LOCK
      See Also:

    • logger

      protected final Log logger

  • Constructor Details

    • AbstractMessageGroupStore

      protected AbstractMessageGroupStore()

    • AbstractMessageGroupStore

      protected AbstractMessageGroupStore(boolean lazyLoadMessageGroups)

  • Method Details

    • getMessageGroupFactory

      protected MessageGroupFactory getMessageGroupFactory()
      Overrides:
      getMessageGroupFactory in class AbstractBatchingMessageGroupStore

    • setExpiryCallbacks

      public void setExpiryCallbacks(Collection<MessageGroupStore.MessageGroupCallback> expiryCallbacks)
      Convenient injection point for expiry callbacks in the message store. Each of the callbacks provided will simply be registered with the store using registerMessageGroupExpiryCallback(MessageGroupCallback).
      Parameters:
      expiryCallbacks - the expiry callbacks to add

    • isTimeoutOnIdle

      public boolean isTimeoutOnIdle()

    • setTimeoutOnIdle

      public void setTimeoutOnIdle(boolean timeoutOnIdle)
      Allows you to override the rule for the timeout calculation. Typical timeout is based from the time the MessageGroup was created. If you want the timeout to be based on the time the MessageGroup was idling (e.g., inactive from the last update) invoke this method with 'true'. Default is 'false'.
      Parameters:
      timeoutOnIdle - The boolean.

    • setLazyLoadMessageGroups

      public void setLazyLoadMessageGroups(boolean lazyLoadMessageGroups)
      Specify if the result of the BasicMessageGroupStore.getMessageGroup(Object) should be wrapped to the PersistentMessageGroup - a lazy-load proxy for messages in group Defaults to true.

      The target logic is based on the SimpleMessageGroupFactory.GroupType.PERSISTENT.

      Parameters:
      lazyLoadMessageGroups - the boolean flag to use.
      Since:
      4.3

    • setLockRegistry

      public final void setLockRegistry(LockRegistry<?> lockRegistry)
      Specify the type of the LockRegistry to ensure atomic operations
      Parameters:
      lockRegistry - lockRegistryType
      Since:
      6.5

    • getLockRegistry

      protected LockRegistry<?> getLockRegistry()

    • registerMessageGroupExpiryCallback

      public void registerMessageGroupExpiryCallback(MessageGroupStore.MessageGroupCallback callback)
      Description copied from interface: MessageGroupStore
      Register a callback for when a message group is expired through MessageGroupStore.expireMessageGroups(long).
      Specified by:
      registerMessageGroupExpiryCallback in interface MessageGroupStore
      Parameters:
      callback - A callback to execute when a message group is cleaned up.

    • expireMessageGroups

      @ManagedOperation public int expireMessageGroups(long timeout)
      Description copied from interface: MessageGroupStore
      Extract all expired groups (whose timestamp is older than the current time less the threshold provided) and call each of the registered callbacks on them in turn. For example: call with a timeout of 100 to expire all groups that were created more than 100 milliseconds ago, and are not yet complete. Use a timeout of 0 (or negative to be on the safe side) to expire all message groups.
      Specified by:
      expireMessageGroups in interface MessageGroupStore
      Parameters:
      timeout - the timeout threshold to use
      Returns:
      the number of message groups expired
      See Also:

    • copy

      protected MessageGroup copy(MessageGroup group)
      Used by expireMessageGroups. We need to return a snapshot of the group at the time the reaper runs, so we can properly detect if the group changed between now and the attempt to expire the group. Not necessary for persistent stores, so the default behavior is to just return the group.
      Parameters:
      group - The group.
      Returns:
      The group, or a copy.
      Since:
      4.0.1

    • getMessageCountForAllMessageGroups

      @ManagedAttribute public int getMessageCountForAllMessageGroups()
      Description copied from interface: MessageGroupStore
      Optional attribute giving the number of messages in the store over all groups. Implementations may decline to respond by throwing an exception.
      Specified by:
      getMessageCountForAllMessageGroups in interface MessageGroupStore
      Returns:
      the number of messages

    • getMessageGroupCount

      @ManagedAttribute public int getMessageGroupCount()
      Description copied from interface: MessageGroupStore
      Optional attribute giving the number of message groups. Implementations may decline to respond by throwing an exception.
      Specified by:
      getMessageGroupCount in interface MessageGroupStore
      Returns:
      the number message groups

    • removeMessagesFromGroup

      public void removeMessagesFromGroup(Object key, Message<?>... messages)
      Description copied from interface: MessageGroupStore
      Persist the deletion of messages from the group.
      Specified by:
      removeMessagesFromGroup in interface MessageGroupStore
      Parameters:
      key - The groupId for the group containing the message(s).
      messages - The messages to be removed.

    • removeMessagesFromGroup

      public void removeMessagesFromGroup(Object key, Collection<Message<?>> messages)
      Description copied from interface: MessageGroupStore
      Persist the deletion of messages from the group.
      Specified by:
      removeMessagesFromGroup in interface MessageGroupStore
      Parameters:
      key - The groupId for the group containing the message(s).
      messages - The messages to be removed.

    • doRemoveMessagesFromGroup

      protected abstract void doRemoveMessagesFromGroup(Object key, Collection<Message<?>> messages)

    • addMessagesToGroup

      public void addMessagesToGroup(Object groupId, Message<?>... messages)
      Description copied from interface: MessageGroupStore
      Store messages with an association to a group id. This can be used to group messages together.
      Specified by:
      addMessagesToGroup in interface MessageGroupStore
      Parameters:
      groupId - The group id to store messages under.
      messages - The messages to add.

    • doAddMessagesToGroup

      protected abstract void doAddMessagesToGroup(Object groupId, Message<?>... messages)

    • addMessageToGroup

      public MessageGroup addMessageToGroup(Object groupId, Message<?> message)
      Description copied from interface: BasicMessageGroupStore
      Store a message with an association to a group id. This can be used to group messages together.
      Specified by:
      addMessageToGroup in interface BasicMessageGroupStore
      Parameters:
      groupId - The group id to store the message under.
      message - A message.
      Returns:
      The message group.

    • removeMessageGroup

      public void removeMessageGroup(Object groupId)
      Description copied from interface: BasicMessageGroupStore
      Remove the message group with this id.
      Specified by:
      removeMessageGroup in interface BasicMessageGroupStore
      Parameters:
      groupId - The id of the group to remove.

    • doRemoveMessageGroup

      protected abstract void doRemoveMessageGroup(Object groupId)

    • removeMessageFromGroupById

      public boolean removeMessageFromGroupById(Object groupId, UUID messageId)
      Description copied from interface: MessageGroupStore
      Deletion the message from the group.
      Specified by:
      removeMessageFromGroupById in interface MessageGroupStore
      Parameters:
      groupId - The groupId for the group containing the message.
      messageId - The message id to be removed.
      Returns:
      true if message has been removed.

    • doRemoveMessageFromGroupById

      protected boolean doRemoveMessageFromGroupById(Object groupId, UUID messageId)

    • setLastReleasedSequenceNumberForGroup

      public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber)
      Description copied from interface: MessageGroupStore
      Allows you to set the sequence number of the last released Message. Used for Resequencing use cases
      Specified by:
      setLastReleasedSequenceNumberForGroup in interface MessageGroupStore
      Parameters:
      groupId - The group identifier.
      sequenceNumber - The sequence number.

    • doSetLastReleasedSequenceNumberForGroup

      protected abstract void doSetLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber)

    • completeGroup

      public void completeGroup(Object groupId)
      Description copied from interface: MessageGroupStore
      Completes this MessageGroup. Completion of the MessageGroup generally means 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.
      Specified by:
      completeGroup in interface MessageGroupStore
      Parameters:
      groupId - The group identifier.

    • doCompleteGroup

      protected abstract void doCompleteGroup(Object groupId)

    • setGroupCondition

      public void setGroupCondition(Object groupId, String condition)
      Description copied from interface: MessageGroupStore
      Add a condition sentence into the group. Can be used later on for making some decisions for group, e.g. release strategy for correlation handler can consult this condition instead of iterating all the messages in group.
      Specified by:
      setGroupCondition in interface MessageGroupStore
      Parameters:
      groupId - The group identifier.
      condition - The condition to store into the group.

    • doSetGroupCondition

      protected abstract void doSetGroupCondition(Object groupId, String condition)

    • pollMessageFromGroup

      public @Nullable Message<?> pollMessageFromGroup(Object groupId)
      Description copied from interface: BasicMessageGroupStore
      Poll Message from this MessageGroup (in FIFO style if supported by the implementation) while also removing the polled Message.
      Specified by:
      pollMessageFromGroup in interface BasicMessageGroupStore
      Parameters:
      groupId - The group identifier.
      Returns:
      The message.

    • doPollMessageFromGroup

      protected abstract @Nullable Message<?> doPollMessageFromGroup(Object groupId)

    • executeLocked

      protected <T, E extends RuntimeException> T executeLocked(Object groupId, CheckedCallable<T,E> runnable)

    • executeLocked

      protected <E extends RuntimeException> void executeLocked(Object groupId, CheckedRunnable<E> runnable)