Class JdbcMessageStore

java.lang.Object
org.springframework.integration.store.AbstractBatchingMessageGroupStore
org.springframework.integration.store.AbstractMessageGroupStore
org.springframework.integration.jdbc.store.JdbcMessageStore
All Implemented Interfaces:
Iterable<org.springframework.integration.store.MessageGroup>, org.springframework.beans.factory.Aware, org.springframework.beans.factory.BeanClassLoaderAware, org.springframework.context.Lifecycle, org.springframework.context.Phased, org.springframework.context.SmartLifecycle, org.springframework.integration.store.BasicMessageGroupStore, org.springframework.integration.store.MessageGroupStore, org.springframework.integration.store.MessageStore
public class JdbcMessageStore extends org.springframework.integration.store.AbstractMessageGroupStore implements org.springframework.integration.store.MessageStore, org.springframework.beans.factory.BeanClassLoaderAware, org.springframework.context.SmartLifecycle
Implementation of MessageStore using a relational database via JDBC. SQL scripts to create the necessary tables are packaged as org/springframework/integration/jdbc/schema-*.sql, where * is the target database type.

If you intend backing a MessageChannel using a JDBC-based Message Store, please consider using the channel-specific JdbcChannelMessageStore instead. This implementation is intended for correlation components (e.g. <aggregator>), <delayer> and similar.

This class implements SmartLifecycle and calls getMessageGroupCount() on start() to check if required tables are present in DB. The application context will fail to start if the table is not present. This check can be disabled via setCheckDatabaseOnStart(boolean).

Since:
2.0

  • Nested Class Summary

    Nested classes/interfaces inherited from interface org.springframework.integration.store.MessageGroupStore

    org.springframework.integration.store.MessageGroupStore.MessageGroupCallback

  • Field Summary

    Fields
    Modifier and Type
    Field
    Description
    static final String
    DEFAULT_TABLE_PREFIX
    Default value for the table prefix property.

    Fields inherited from class org.springframework.integration.store.AbstractMessageGroupStore

    INTERRUPTED_WHILE_OBTAINING_LOCK, logger

    Fields inherited from interface org.springframework.context.SmartLifecycle

    DEFAULT_PHASE

  • Constructor Summary

    Constructors
    Constructor
    Description
    JdbcMessageStore(DataSource dataSource)
    Create a MessageStore with all mandatory properties.
    JdbcMessageStore(org.springframework.jdbc.core.JdbcOperations jdbcOperations)
    Create a MessageStore with all mandatory properties.

  • Method Summary

    Modifier and Type
    Method
    Description
    void
    addAllowedPatterns(String... patterns)
    Add patterns for packages/classes that are allowed to be deserialized.
    <T> org.springframework.messaging.Message<T>
    addMessage(org.springframework.messaging.Message<T> message)
     
    protected void
    doAddMessagesToGroup(Object groupId, org.springframework.messaging.Message<?>... messages)
     
    protected void
    doCompleteGroup(Object groupId)
     
    protected @Nullable org.springframework.messaging.Message<?>
    doPollForMessage(String groupIdKey)
    This method executes a call to the DB to get the oldest Message in the MessageGroup Override this method if need to.
    protected @Nullable org.springframework.messaging.Message<?>
    doPollMessageFromGroup(Object groupId)
     
    protected boolean
    doRemoveMessageFromGroupById(Object groupId, UUID messageId)
     
    protected void
    doRemoveMessageGroup(Object groupId)
     
    protected void
    doRemoveMessagesFromGroup(Object groupId, Collection<org.springframework.messaging.Message<?>> messages)
     
    protected void
    doSetGroupCondition(Object groupId, String condition)
     
    protected void
    doSetLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber)
     
    @Nullable org.springframework.integration.store.MessageGroupMetadata
    getGroupMetadata(Object groupId)
     
    protected org.springframework.jdbc.core.JdbcOperations
    getJdbcOperations()
    To be used to get a reference to JdbcOperations in case this class is subclassed.
    @Nullable org.springframework.messaging.Message<?>
    getMessage(UUID id)
     
    long
    getMessageCount()
     
    int
    getMessageCountForAllMessageGroups()
     
    @Nullable org.springframework.messaging.Message<?>
    getMessageFromGroup(Object groupId, UUID messageId)
     
    org.springframework.integration.store.MessageGroup
    getMessageGroup(Object groupId)
     
    int
    getMessageGroupCount()
     
    @Nullable org.springframework.integration.store.MessageMetadata
    getMessageMetadata(UUID id)
     
    Collection<org.springframework.messaging.Message<?>>
    getMessagesForGroup(Object groupId)
     
    @Nullable org.springframework.messaging.Message<?>
    getOneMessageFromGroup(Object groupId)
     
    protected String
    getQuery(org.springframework.integration.jdbc.store.JdbcMessageStore.Query base)
    Replace patterns in the input to produce a valid SQL query.
    boolean
    isAutoStartup()
     
    boolean
    isRunning()
     
    Iterator<org.springframework.integration.store.MessageGroup>
    iterator()
     
    int
    messageGroupSize(Object groupId)
     
    @Nullable org.springframework.messaging.Message<?>
    removeMessage(UUID id)
     
    void
    setBeanClassLoader(ClassLoader classLoader)
     
    void
    setCheckDatabaseOnStart(boolean checkDatabaseOnStart)
    The flag to perform a database check query on start or not.
    void
    setDeserializer(org.springframework.core.serializer.Deserializer<? extends org.springframework.messaging.Message<?>> deserializer)
    A converter for deserializing byte arrays to message.
    void
    setRegion(String region)
    A unique grouping identifier for all messages persisted with this store.
    void
    setSerializer(org.springframework.core.serializer.Serializer<? super org.springframework.messaging.Message<?>> serializer)
    A converter for serializing messages to byte arrays for storage.
    void
    setTablePrefix(String tablePrefix)
    Public setter for the table prefix property.
    void
    start()
     
    void
    stop()
     
    Stream<org.springframework.messaging.Message<?>>
    streamMessagesForGroup(Object groupId)
     

    Methods inherited from class org.springframework.integration.store.AbstractMessageGroupStore

    addMessagesToGroup, addMessageToGroup, completeGroup, copy, executeLocked, executeLocked, expireMessageGroups, getLockRegistry, getMessageGroupFactory, isTimeoutOnIdle, pollMessageFromGroup, registerMessageGroupExpiryCallback, removeMessageFromGroupById, removeMessageGroup, removeMessagesFromGroup, removeMessagesFromGroup, setExpiryCallbacks, setGroupCondition, setLastReleasedSequenceNumberForGroup, setLazyLoadMessageGroups, setLockRegistry, setTimeoutOnIdle

    Methods inherited from class org.springframework.integration.store.AbstractBatchingMessageGroupStore

    getRemoveBatchSize, setMessageGroupFactory, setRemoveBatchSize

    Methods inherited from class Object

    clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

    Methods inherited from interface Iterable

    forEach, spliterator

    Methods inherited from interface org.springframework.context.SmartLifecycle

    getPhase, isPauseable, stop

  • Field Details

    • DEFAULT_TABLE_PREFIX

      public static final String DEFAULT_TABLE_PREFIX
      Default value for the table prefix property.
      See Also:

  • Constructor Details

    • JdbcMessageStore

      public JdbcMessageStore(DataSource dataSource)
      Create a MessageStore with all mandatory properties.
      Parameters:
      dataSource - a DataSource

    • JdbcMessageStore

      public JdbcMessageStore(org.springframework.jdbc.core.JdbcOperations jdbcOperations)
      Create a MessageStore with all mandatory properties.
      Parameters:
      jdbcOperations - a JdbcOperations
      Since:
      4.3.9

  • Method Details

    • setBeanClassLoader

      public void setBeanClassLoader(ClassLoader classLoader)
      Specified by:
      setBeanClassLoader in interface org.springframework.beans.factory.BeanClassLoaderAware

    • setTablePrefix

      public void setTablePrefix(String tablePrefix)
      Public setter for the table prefix property. This will be prefixed to all the table names before queries are executed. Defaults to DEFAULT_TABLE_PREFIX.
      Parameters:
      tablePrefix - the tablePrefix to set

    • setRegion

      public void setRegion(String region)
      A unique grouping identifier for all messages persisted with this store. Using multiple regions allows the store to be partitioned (if necessary) for different purposes. Defaults to DEFAULT.
      Parameters:
      region - the region name to set

    • setSerializer

      public void setSerializer(org.springframework.core.serializer.Serializer<? super org.springframework.messaging.Message<?>> serializer)
      A converter for serializing messages to byte arrays for storage.
      Parameters:
      serializer - the serializer to set

    • setDeserializer

      public void setDeserializer(org.springframework.core.serializer.Deserializer<? extends org.springframework.messaging.Message<?>> deserializer)
      A converter for deserializing byte arrays to message.
      Parameters:
      deserializer - the deserializer to set

    • addAllowedPatterns

      public void addAllowedPatterns(String... patterns)
      Add patterns for packages/classes that are allowed to be deserialized. A class can be fully qualified or a wildcard '*' is allowed at the beginning or end of the class name. Examples: com.foo.*, *.MyClass.
      Parameters:
      patterns - the patterns.
      Since:
      5.4

    • setCheckDatabaseOnStart

      public void setCheckDatabaseOnStart(boolean checkDatabaseOnStart)
      The flag to perform a database check query on start or not.
      Parameters:
      checkDatabaseOnStart - false to not perform the database check.
      Since:
      6.2

    • isAutoStartup

      public boolean isAutoStartup()
      Specified by:
      isAutoStartup in interface org.springframework.context.SmartLifecycle

    • start

      public void start()
      Specified by:
      start in interface org.springframework.context.Lifecycle

    • stop

      public void stop()
      Specified by:
      stop in interface org.springframework.context.Lifecycle

    • isRunning

      public boolean isRunning()
      Specified by:
      isRunning in interface org.springframework.context.Lifecycle

    • removeMessage

      public @Nullable org.springframework.messaging.Message<?> removeMessage(UUID id)
      Specified by:
      removeMessage in interface org.springframework.integration.store.MessageStore

    • getMessageCount

      @ManagedAttribute public long getMessageCount()
      Specified by:
      getMessageCount in interface org.springframework.integration.store.MessageStore

    • getMessage

      public @Nullable org.springframework.messaging.Message<?> getMessage(UUID id)
      Specified by:
      getMessage in interface org.springframework.integration.store.MessageStore

    • getMessageMetadata

      public @Nullable org.springframework.integration.store.MessageMetadata getMessageMetadata(UUID id)
      Specified by:
      getMessageMetadata in interface org.springframework.integration.store.MessageStore

    • addMessage

      public <T> org.springframework.messaging.Message<T> addMessage(org.springframework.messaging.Message<T> message)
      Specified by:
      addMessage in interface org.springframework.integration.store.MessageStore

    • doAddMessagesToGroup

      protected void doAddMessagesToGroup(Object groupId, org.springframework.messaging.Message<?>... messages)
      Specified by:
      doAddMessagesToGroup in class org.springframework.integration.store.AbstractMessageGroupStore

    • getMessageGroupCount

      @ManagedAttribute public int getMessageGroupCount()
      Specified by:
      getMessageGroupCount in interface org.springframework.integration.store.MessageGroupStore
      Overrides:
      getMessageGroupCount in class org.springframework.integration.store.AbstractMessageGroupStore

    • getMessageCountForAllMessageGroups

      @ManagedAttribute public int getMessageCountForAllMessageGroups()
      Specified by:
      getMessageCountForAllMessageGroups in interface org.springframework.integration.store.MessageGroupStore
      Overrides:
      getMessageCountForAllMessageGroups in class org.springframework.integration.store.AbstractMessageGroupStore

    • messageGroupSize

      @ManagedAttribute public int messageGroupSize(Object groupId)
      Specified by:
      messageGroupSize in interface org.springframework.integration.store.BasicMessageGroupStore

    • getMessageGroup

      public org.springframework.integration.store.MessageGroup getMessageGroup(Object groupId)
      Specified by:
      getMessageGroup in interface org.springframework.integration.store.BasicMessageGroupStore

    • getGroupMetadata

      public @Nullable org.springframework.integration.store.MessageGroupMetadata getGroupMetadata(Object groupId)
      Specified by:
      getGroupMetadata in interface org.springframework.integration.store.MessageGroupStore

    • doRemoveMessagesFromGroup

      protected void doRemoveMessagesFromGroup(Object groupId, Collection<org.springframework.messaging.Message<?>> messages)
      Specified by:
      doRemoveMessagesFromGroup in class org.springframework.integration.store.AbstractMessageGroupStore

    • getMessageFromGroup

      public @Nullable org.springframework.messaging.Message<?> getMessageFromGroup(Object groupId, UUID messageId)
      Specified by:
      getMessageFromGroup in interface org.springframework.integration.store.MessageGroupStore

    • doRemoveMessageFromGroupById

      protected boolean doRemoveMessageFromGroupById(Object groupId, UUID messageId)
      Overrides:
      doRemoveMessageFromGroupById in class org.springframework.integration.store.AbstractMessageGroupStore

    • doRemoveMessageGroup

      protected void doRemoveMessageGroup(Object groupId)
      Specified by:
      doRemoveMessageGroup in class org.springframework.integration.store.AbstractMessageGroupStore

    • doCompleteGroup

      protected void doCompleteGroup(Object groupId)
      Specified by:
      doCompleteGroup in class org.springframework.integration.store.AbstractMessageGroupStore

    • doSetGroupCondition

      protected void doSetGroupCondition(Object groupId, String condition)
      Specified by:
      doSetGroupCondition in class org.springframework.integration.store.AbstractMessageGroupStore

    • doSetLastReleasedSequenceNumberForGroup

      protected void doSetLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber)
      Specified by:
      doSetLastReleasedSequenceNumberForGroup in class org.springframework.integration.store.AbstractMessageGroupStore

    • doPollMessageFromGroup

      protected @Nullable org.springframework.messaging.Message<?> doPollMessageFromGroup(Object groupId)
      Specified by:
      doPollMessageFromGroup in class org.springframework.integration.store.AbstractMessageGroupStore

    • getOneMessageFromGroup

      public @Nullable org.springframework.messaging.Message<?> getOneMessageFromGroup(Object groupId)
      Specified by:
      getOneMessageFromGroup in interface org.springframework.integration.store.MessageGroupStore

    • getMessagesForGroup

      public Collection<org.springframework.messaging.Message<?>> getMessagesForGroup(Object groupId)
      Specified by:
      getMessagesForGroup in interface org.springframework.integration.store.MessageGroupStore

    • streamMessagesForGroup

      public Stream<org.springframework.messaging.Message<?>> streamMessagesForGroup(Object groupId)
      Specified by:
      streamMessagesForGroup in interface org.springframework.integration.store.MessageGroupStore

    • iterator

      public Iterator<org.springframework.integration.store.MessageGroup> iterator()
      Specified by:
      iterator in interface Iterable<org.springframework.integration.store.MessageGroup>
      Specified by:
      iterator in interface org.springframework.integration.store.MessageGroupStore

    • getQuery

      protected String getQuery(org.springframework.integration.jdbc.store.JdbcMessageStore.Query base)
      Replace patterns in the input to produce a valid SQL query. This implementation lazily initializes a simple map-based cache, only replacing the table prefix on the first access to a named query. Further, accesses will be resolved from the cache.
      Parameters:
      base - the SQL query to be transformed
      Returns:
      a transformed query with replacements

    • getJdbcOperations

      protected org.springframework.jdbc.core.JdbcOperations getJdbcOperations()
      To be used to get a reference to JdbcOperations in case this class is subclassed.
      Returns:
      the JdbcOperations implementation

    • doPollForMessage

      protected @Nullable org.springframework.messaging.Message<?> doPollForMessage(String groupIdKey)
      This method executes a call to the DB to get the oldest Message in the MessageGroup Override this method if need to. For example if your DB supports advanced function such as FIRST etc.
      Parameters:
      groupIdKey - String representation of message group ID
      Returns:
      a message; could be null if query produced no Messages