INT-4081: Post-merge polishing and improvements
JIRA: https://jira.spring.io/browse/INT-4081 * Rename `MessageGroupPreparedStatementSetter` to `ChannelMessageStorePreparedStatementSetter` * Add JavaDocs to the `ChannelMessageStorePreparedStatementSetter` * Polishing `JdbcChannelMessageStore` JavaDocs; remove unused `BeanFactory` * Polishing Docs for typos and proper text flow
This commit is contained in:
Artem Bilan
@@ -32,16 +32,13 @@ import javax.sql.DataSource;
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
import org.springframework.beans.BeansException;
|
||||
import org.springframework.beans.factory.BeanFactory;
|
||||
import org.springframework.beans.factory.BeanFactoryAware;
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.core.serializer.Deserializer;
|
||||
import org.springframework.core.serializer.Serializer;
|
||||
import org.springframework.core.serializer.support.SerializingConverter;
|
||||
import org.springframework.dao.DuplicateKeyException;
|
||||
import org.springframework.integration.jdbc.store.channel.ChannelMessageStorePreparedStatementSetter;
|
||||
import org.springframework.integration.jdbc.store.channel.ChannelMessageStoreQueryProvider;
|
||||
import org.springframework.integration.jdbc.store.channel.MessageGroupPreparedStatementSetter;
|
||||
import org.springframework.integration.jdbc.store.channel.MessageRowMapper;
|
||||
import org.springframework.integration.jdbc.store.channel.OracleChannelMessageStoreQueryProvider;
|
||||
import org.springframework.integration.store.MessageGroup;
|
||||
@@ -88,26 +85,14 @@ import org.springframework.util.StringUtils;
|
||||
* @author Artem Bilan
|
||||
* @author Gary Russell
|
||||
* @author Meherzad Lahewala
|
||||
*
|
||||
* @since 2.2
|
||||
*/
|
||||
@ManagedResource
|
||||
public class JdbcChannelMessageStore implements PriorityCapableChannelMessageStore, InitializingBean, BeanFactoryAware {
|
||||
public class JdbcChannelMessageStore implements PriorityCapableChannelMessageStore, InitializingBean {
|
||||
|
||||
private static final Log logger = LogFactory.getLog(JdbcChannelMessageStore.class);
|
||||
|
||||
private final Set<String> idCache = new HashSet<String>();
|
||||
|
||||
private final ReadWriteLock idCacheLock = new ReentrantReadWriteLock();
|
||||
|
||||
private final Lock idCacheReadLock = this.idCacheLock.readLock();
|
||||
|
||||
private final Lock idCacheWriteLock = this.idCacheLock.writeLock();
|
||||
|
||||
/**
|
||||
* Default value for the table prefix property.
|
||||
*/
|
||||
public static final String DEFAULT_TABLE_PREFIX = "INT_";
|
||||
|
||||
/**
|
||||
* Default region property, used to partition the message store. For example,
|
||||
* a separate Spring Integration application with overlapping channel names
|
||||
@@ -115,7 +100,10 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
*/
|
||||
public static final String DEFAULT_REGION = "DEFAULT";
|
||||
|
||||
private ChannelMessageStoreQueryProvider channelMessageStoreQueryProvider;
|
||||
/**
|
||||
* Default value for the table prefix property.
|
||||
*/
|
||||
public static final String DEFAULT_TABLE_PREFIX = "INT_";
|
||||
|
||||
/**
|
||||
* The name of the message header that stores a flag to indicate that the message has been saved. This is an
|
||||
@@ -132,33 +120,40 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
@Deprecated
|
||||
public static final String CREATED_DATE_KEY = JdbcChannelMessageStore.class.getSimpleName() + ".CREATED_DATE";
|
||||
|
||||
private volatile String region = DEFAULT_REGION;
|
||||
private final Set<String> idCache = new HashSet<>();
|
||||
|
||||
private volatile String tablePrefix = DEFAULT_TABLE_PREFIX;
|
||||
private final ReadWriteLock idCacheLock = new ReentrantReadWriteLock();
|
||||
|
||||
private volatile JdbcTemplate jdbcTemplate;
|
||||
private final Lock idCacheReadLock = this.idCacheLock.readLock();
|
||||
|
||||
private volatile WhiteListDeserializingConverter deserializer;
|
||||
private final Lock idCacheWriteLock = this.idCacheLock.writeLock();
|
||||
|
||||
private volatile SerializingConverter serializer;
|
||||
private ChannelMessageStoreQueryProvider channelMessageStoreQueryProvider;
|
||||
|
||||
private volatile LobHandler lobHandler = new DefaultLobHandler();
|
||||
private String region = DEFAULT_REGION;
|
||||
|
||||
private volatile MessageRowMapper messageRowMapper;
|
||||
private String tablePrefix = DEFAULT_TABLE_PREFIX;
|
||||
|
||||
private volatile MessageGroupPreparedStatementSetter messageGroupPreparedStatementSetter;
|
||||
private JdbcTemplate jdbcTemplate;
|
||||
|
||||
private volatile Map<String, String> queryCache = new HashMap<String, String>();
|
||||
private WhiteListDeserializingConverter deserializer;
|
||||
|
||||
private volatile MessageGroupFactory messageGroupFactory = new SimpleMessageGroupFactory();
|
||||
private SerializingConverter serializer;
|
||||
|
||||
private LobHandler lobHandler = new DefaultLobHandler();
|
||||
|
||||
private MessageRowMapper messageRowMapper;
|
||||
|
||||
private ChannelMessageStorePreparedStatementSetter preparedStatementSetter;
|
||||
|
||||
private Map<String, String> queryCache = new HashMap<>();
|
||||
|
||||
private MessageGroupFactory messageGroupFactory = new SimpleMessageGroupFactory();
|
||||
|
||||
private boolean usingIdCache = false;
|
||||
|
||||
private boolean priorityEnabled;
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
private BeanFactory beanFactory;
|
||||
|
||||
/**
|
||||
* Convenient constructor for configuration use.
|
||||
*/
|
||||
@@ -170,16 +165,13 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
/**
|
||||
* Create a {@link MessageStore} with all mandatory properties. The passed-in
|
||||
* {@link DataSource} is used to instantiate a {@link JdbcTemplate}
|
||||
*
|
||||
* with {@link JdbcTemplate#setFetchSize(int)} set to <code>1</code>
|
||||
* and with {@link JdbcTemplate#setMaxRows(int)} set to <code>1</code>.
|
||||
*
|
||||
* @param dataSource a {@link DataSource}
|
||||
*/
|
||||
public JdbcChannelMessageStore(DataSource dataSource) {
|
||||
this();
|
||||
this.jdbcTemplate = new JdbcTemplate(dataSource);
|
||||
|
||||
this.jdbcTemplate.setFetchSize(1);
|
||||
this.jdbcTemplate.setMaxRows(1);
|
||||
}
|
||||
@@ -189,22 +181,19 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
* The passed-in {@link DataSource} is used to instantiate a {@link JdbcTemplate}
|
||||
* with {@link JdbcTemplate#setFetchSize(int)} set to <code>1</code>
|
||||
* and with {@link JdbcTemplate#setMaxRows(int)} set to <code>1</code>.
|
||||
*
|
||||
* @param dataSource a {@link DataSource}
|
||||
*/
|
||||
public void setDataSource(DataSource dataSource) {
|
||||
this.jdbcTemplate = new JdbcTemplate(dataSource);
|
||||
|
||||
this.jdbcTemplate.setFetchSize(1);
|
||||
this.jdbcTemplate.setMaxRows(1);
|
||||
}
|
||||
|
||||
/**
|
||||
* A converter for deserializing byte arrays to messages.
|
||||
*
|
||||
* @param deserializer the deserializer to set
|
||||
*/
|
||||
@SuppressWarnings({"unchecked", "rawtypes"})
|
||||
@SuppressWarnings({ "unchecked", "rawtypes" })
|
||||
public void setDeserializer(Deserializer<? extends Message<?>> deserializer) {
|
||||
this.deserializer = new WhiteListDeserializingConverter((Deserializer) deserializer);
|
||||
}
|
||||
@@ -223,11 +212,9 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
/**
|
||||
* The {@link JdbcOperations} to use when interacting with the database. Either
|
||||
* this property can be set or the {@link #setDataSource(DataSource) dataSource}.
|
||||
*
|
||||
* Please consider passing in a {@link JdbcTemplate} with a fetchSize property
|
||||
* of 1. This is particularly important for Oracle to ensure First In, First Out (FIFO)
|
||||
* message retrieval characteristics.
|
||||
*
|
||||
* @param jdbcTemplate a {@link JdbcOperations}
|
||||
*/
|
||||
public void setJdbcTemplate(JdbcTemplate jdbcTemplate) {
|
||||
@@ -238,7 +225,6 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
/**
|
||||
* Override the {@link LobHandler} that is used to create and unpack large objects in SQL queries. The default is
|
||||
* fine for almost all platforms, but some Oracle drivers require a native implementation.
|
||||
*
|
||||
* @param lobHandler a {@link LobHandler}
|
||||
*/
|
||||
public void setLobHandler(LobHandler lobHandler) {
|
||||
@@ -250,7 +236,6 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
* Allows for passing in a custom {@link MessageRowMapper}. The {@link MessageRowMapper}
|
||||
* is used to convert the selected database row representing the persisted
|
||||
* message into the actual {@link Message} object.
|
||||
*
|
||||
* @param messageRowMapper Must not be null
|
||||
*/
|
||||
public void setMessageRowMapper(MessageRowMapper messageRowMapper) {
|
||||
@@ -259,17 +244,15 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
}
|
||||
|
||||
/**
|
||||
* Allows for passing in a custom {@link MessageGroupPreparedStatementSetter}.
|
||||
* The {@link MessageGroupPreparedStatementSetter} is used to insert message into the database.
|
||||
*
|
||||
* @param messageGroupPreparedStatementSetter Must not be null
|
||||
* Set a {@link ChannelMessageStorePreparedStatementSetter} to insert message into the database.
|
||||
* @param preparedStatementSetter {@link ChannelMessageStorePreparedStatementSetter} to use.
|
||||
* Must not be null
|
||||
* @since 5.0
|
||||
*/
|
||||
public void setMessageGroupPreparedStatementSetter(
|
||||
MessageGroupPreparedStatementSetter messageGroupPreparedStatementSetter) {
|
||||
Assert.notNull(messageGroupPreparedStatementSetter,
|
||||
"The provided MessageGroupPreparedStatementSetter must not be null.");
|
||||
this.messageGroupPreparedStatementSetter = messageGroupPreparedStatementSetter;
|
||||
public void setPreparedStatementSetter(ChannelMessageStorePreparedStatementSetter preparedStatementSetter) {
|
||||
Assert.notNull(preparedStatementSetter,
|
||||
"The provided ChannelMessageStorePreparedStatementSetter must not be null.");
|
||||
this.preparedStatementSetter = preparedStatementSetter;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -291,7 +274,6 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
* 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 <code>{@link #DEFAULT_REGION}</code>.
|
||||
*
|
||||
* @param region the region name to set
|
||||
*/
|
||||
public void setRegion(String region) {
|
||||
@@ -300,7 +282,6 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
|
||||
/**
|
||||
* A converter for serializing messages to byte arrays for storage.
|
||||
*
|
||||
* @param serializer The serializer to set
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
@@ -312,7 +293,6 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
/**
|
||||
* Public setter for the table prefix property. This will be prefixed to all the table names before queries are
|
||||
* executed. Defaults to {@link #DEFAULT_TABLE_PREFIX}.
|
||||
*
|
||||
* @param tablePrefix the tablePrefix to set
|
||||
*/
|
||||
public void setTablePrefix(String tablePrefix) {
|
||||
@@ -323,23 +303,18 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
* <p>Consider using this property when polling the database transactionally
|
||||
* using multiple parallel threads, meaning when the configured poller is configured
|
||||
* using a task executor.</p>
|
||||
*
|
||||
* <p>The issue is that the {@link #pollMessageFromGroup(Object)} looks for the
|
||||
* oldest entry for a giving channel (groupKey) and region ({@link #setRegion(String)}).
|
||||
* If you do that with multiple threads and you are using transactions, other
|
||||
* threads may be waiting for that same locked row.</p>
|
||||
*
|
||||
* <p>If using the provided {@link OracleChannelMessageStoreQueryProvider}, don't set {@link #usingIdCache}
|
||||
* to true, as the Oracle query will ignore locked rows.</p>
|
||||
*
|
||||
* <p>Using the id cache, the {@link JdbcChannelMessageStore} will store each
|
||||
* message id in an in-memory collection for the duration of processing. With
|
||||
* that, any polling threads will explicitly exclude those messages from
|
||||
* being polled.</p>
|
||||
*
|
||||
* <p>For this to work, you must setup the corresponding
|
||||
* {@link TransactionSynchronizationFactory}:</p>
|
||||
*
|
||||
* <pre class="code">
|
||||
* {@code
|
||||
* <int:transaction-synchronization-factory id="syncFactory">
|
||||
@@ -348,10 +323,8 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
* </int:transaction-synchronization-factory>
|
||||
* }
|
||||
* </pre>
|
||||
*
|
||||
* This {@link TransactionSynchronizationFactory} is then referenced in the
|
||||
* transaction configuration of the poller:
|
||||
*
|
||||
* <pre class="code">
|
||||
* {@code
|
||||
* <int:poller fixed-delay="300" receive-timeout="500"
|
||||
@@ -361,7 +334,6 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
* </int:poller>
|
||||
* }
|
||||
* </pre>
|
||||
*
|
||||
* @param usingIdCache When <code>true</code> the id cache will be used.
|
||||
*/
|
||||
public void setUsingIdCache(boolean usingIdCache) {
|
||||
@@ -393,25 +365,18 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
return this.messageGroupFactory;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
|
||||
this.beanFactory = beanFactory;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check mandatory properties ({@link DataSource} and
|
||||
* {@link #setChannelMessageStoreQueryProvider(ChannelMessageStoreQueryProvider)}). If no {@link MessageRowMapper}
|
||||
* and {@link MessageGroupPreparedStatementSetter} was explicitly set using
|
||||
* and {@link ChannelMessageStorePreparedStatementSetter} was explicitly set using
|
||||
* {@link #setMessageRowMapper(MessageRowMapper)} and
|
||||
* {@link #setMessageGroupPreparedStatementSetter(MessageGroupPreparedStatementSetter)} respectively, the default
|
||||
* {@link MessageRowMapper} and {@link MessageGroupPreparedStatementSetter} will be instantiate using the
|
||||
* {@link #setPreparedStatementSetter(ChannelMessageStorePreparedStatementSetter)} respectively, the default
|
||||
* {@link MessageRowMapper} and {@link ChannelMessageStorePreparedStatementSetter} will be instantiate using the
|
||||
* specified {@link #deserializer} and {@link #lobHandler}.
|
||||
*
|
||||
* Also, if the jdbcTemplate's fetchSize property ({@link JdbcTemplate#getFetchSize()})
|
||||
* is not 1, a warning will be logged. When using the {@link JdbcChannelMessageStore}
|
||||
* with Oracle, the fetchSize value of 1 is needed to ensure FIFO characteristics
|
||||
* of polled messages. Please see the Oracle {@link ChannelMessageStoreQueryProvider} for more details.
|
||||
*
|
||||
* @throws Exception Any Exception.
|
||||
*/
|
||||
@Override
|
||||
@@ -427,8 +392,8 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
logger.warn("The jdbcTemplate's fetch size is not 1. This may cause FIFO issues with Oracle databases.");
|
||||
}
|
||||
|
||||
if (this.messageGroupPreparedStatementSetter == null) {
|
||||
this.messageGroupPreparedStatementSetter = new MessageGroupPreparedStatementSetter(this.serializer,
|
||||
if (this.preparedStatementSetter == null) {
|
||||
this.preparedStatementSetter = new ChannelMessageStorePreparedStatementSetter(this.serializer,
|
||||
this.lobHandler);
|
||||
}
|
||||
this.jdbcTemplate.afterPropertiesSet();
|
||||
@@ -437,9 +402,7 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
/**
|
||||
* Store a message in the database. The groupId identifies the channel for which
|
||||
* the message is to be stored.
|
||||
*
|
||||
* Keep in mind that the actual groupId (Channel Identifier) is converted to a String-based UUID identifier.
|
||||
*
|
||||
* @param groupId the group id to store the message under
|
||||
* @param message a message
|
||||
*/
|
||||
@@ -447,7 +410,7 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
public MessageGroup addMessageToGroup(Object groupId, final Message<?> message) {
|
||||
try {
|
||||
this.jdbcTemplate.update(getQuery(this.channelMessageStoreQueryProvider.getCreateMessageQuery()),
|
||||
ps -> this.messageGroupPreparedStatementSetter.setValues(ps, message, groupId, this.region,
|
||||
ps -> this.preparedStatementSetter.setValues(ps, message, groupId, this.region,
|
||||
this.priorityEnabled));
|
||||
}
|
||||
catch (DuplicateKeyException e) {
|
||||
@@ -462,7 +425,6 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
/**
|
||||
* Helper method that converts the channel id to a UUID using
|
||||
* {@link UUIDConverter#getUUID(Object)}.
|
||||
*
|
||||
* @param input Parameter may be null
|
||||
* @return Returns null when the input is null otherwise the UUID as String.
|
||||
*/
|
||||
@@ -480,13 +442,13 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
|
||||
/**
|
||||
* Method not implemented.
|
||||
*
|
||||
* @return The message group count.
|
||||
* @throws UnsupportedOperationException Method not supported.
|
||||
*/
|
||||
@ManagedAttribute
|
||||
public int getMessageGroupCount() {
|
||||
return this.jdbcTemplate.queryForObject(this.getQuery("SELECT COUNT(DISTINCT GROUP_KEY) from %PREFIX%CHANNEL_MESSAGE where REGION = ?"),
|
||||
return this.jdbcTemplate.queryForObject(
|
||||
getQuery("SELECT COUNT(DISTINCT GROUP_KEY) from %PREFIX%CHANNEL_MESSAGE where REGION = ?"),
|
||||
Integer.class, this.region);
|
||||
}
|
||||
|
||||
@@ -494,7 +456,6 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
* 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.
|
||||
*
|
||||
* @param sqlQuery The SQL query to be transformed.
|
||||
* @return A transformed query with replacements.
|
||||
*/
|
||||
@@ -512,7 +473,6 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
/**
|
||||
* Returns the number of messages persisted for the specified channel id (groupId)
|
||||
* and the specified region ({@link #setRegion(String)}).
|
||||
*
|
||||
* @return The message group size.
|
||||
*/
|
||||
@Override
|
||||
@@ -552,7 +512,6 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
* This method executes a call to the DB to get the oldest Message in the
|
||||
* MessageGroup which in the context of the {@link JdbcChannelMessageStore}
|
||||
* means the channel identifier.
|
||||
*
|
||||
* @param groupIdKey String representation of message group (Channel) ID
|
||||
* @return a message; could be null if query produced no Messages
|
||||
*/
|
||||
@@ -624,7 +583,8 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
final UUID id = messageToRemove.getHeaders().getId();
|
||||
|
||||
int updated = this.jdbcTemplate.update(getQuery(this.channelMessageStoreQueryProvider.getDeleteMessageQuery()),
|
||||
new Object[]{getKey(id), getKey(groupId), this.region}, new int[]{Types.VARCHAR, Types.VARCHAR, Types.VARCHAR});
|
||||
new Object[] { getKey(id), getKey(groupId), this.region },
|
||||
new int[] { Types.VARCHAR, Types.VARCHAR, Types.VARCHAR });
|
||||
|
||||
boolean result = updated != 0;
|
||||
if (result) {
|
||||
@@ -644,7 +604,6 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
* rolled back.</p>
|
||||
* <p>Only applicable if {@link #setUsingIdCache(boolean)} is set to
|
||||
* <code>true</code></p>.
|
||||
*
|
||||
* @param messageId The message identifier.
|
||||
*/
|
||||
public void removeFromIdCache(String messageId) {
|
||||
@@ -663,7 +622,6 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
|
||||
/**
|
||||
* Returns the size of the Message Id Cache, which caches Message Ids for
|
||||
* those messages that are currently being processed.
|
||||
*
|
||||
* @return The size of the Message Id Cache
|
||||
*/
|
||||
@ManagedMetric
|
||||
|
||||
@@ -0,0 +1,135 @@
|
||||
/*
|
||||
* Copyright 2017 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.integration.jdbc.store.channel;
|
||||
|
||||
import java.sql.PreparedStatement;
|
||||
import java.sql.SQLException;
|
||||
import java.sql.Types;
|
||||
import java.util.Objects;
|
||||
|
||||
import org.springframework.core.serializer.support.SerializingConverter;
|
||||
import org.springframework.integration.IntegrationMessageHeaderAccessor;
|
||||
import org.springframework.integration.util.UUIDConverter;
|
||||
import org.springframework.jdbc.support.lob.LobHandler;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* Callback to be used with the {@link org.springframework.integration.jdbc.store.JdbcChannelMessageStore}.
|
||||
* <p>
|
||||
* Behavior is same as standard {@link org.springframework.jdbc.core.PreparedStatementSetter},
|
||||
* it takes in additional {@code Message<?> requestMessage}, {@code Object groupId},
|
||||
* {@code String region} and {@code boolean priorityEnabled} parameters used
|
||||
* for {@code addMessageToGroup} method
|
||||
* in the {@link org.springframework.integration.jdbc.store.JdbcChannelMessageStore}.
|
||||
* <p>
|
||||
* This class can be extended for any custom data structure or columns types.
|
||||
* For this purpose the {@code protected} constructor is provided for inheritors.
|
||||
* In this case the {@link #serializer} and {@link #lobHandler} are null to avoid
|
||||
* extra serialization actions if the target custom behavior doesn't imply them.
|
||||
*
|
||||
* @author Meherzad Lahewala
|
||||
* @author Artem Bilan
|
||||
*
|
||||
* @since 5.0
|
||||
*
|
||||
* @see org.springframework.jdbc.core.PreparedStatementSetter
|
||||
*/
|
||||
public class ChannelMessageStorePreparedStatementSetter {
|
||||
|
||||
private final SerializingConverter serializer;
|
||||
|
||||
private final LobHandler lobHandler;
|
||||
|
||||
/**
|
||||
* Instantiate a {@link ChannelMessageStorePreparedStatementSetter} with the provided
|
||||
* serializer and lobHandler, which both must not be null.
|
||||
* @param serializer the {@link SerializingConverter} to build {@code byte[]} from
|
||||
* the request message
|
||||
* @param lobHandler the {@link LobHandler} to store {@code byte[]} of the request
|
||||
* message to prepared statement
|
||||
*/
|
||||
public ChannelMessageStorePreparedStatementSetter(SerializingConverter serializer, LobHandler lobHandler) {
|
||||
Assert.notNull(serializer, "'serializer' must not be null");
|
||||
Assert.notNull(lobHandler, "'lobHandler' must not be null");
|
||||
this.serializer = serializer;
|
||||
this.lobHandler = lobHandler;
|
||||
}
|
||||
|
||||
/**
|
||||
* The default constructor for inheritors who are not interested in the message
|
||||
* serialization to {@code byte[]}.
|
||||
* The {@link #serializer} and {@link #lobHandler} are null from this constructor,
|
||||
* therefore any serialization isn't happened in the default {@link #setValues} implementation.
|
||||
* A target implementor must ensure the proper custom logic for storing message.
|
||||
*/
|
||||
protected ChannelMessageStorePreparedStatementSetter() {
|
||||
this.serializer = null;
|
||||
this.lobHandler = null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Perform a preparedStatement parameters population according provided arguments.
|
||||
* The default functionality is (parameter - data):
|
||||
* <p>
|
||||
* <ul>
|
||||
* <li>1 - messageId
|
||||
* <li>2 - groupKey
|
||||
* <li>3 - region
|
||||
* <li>4 - createdDate
|
||||
* <li>5 - priority if enabled, otherwise null
|
||||
* <li>6 - serialized message if {@link #serializer} and {@link #lobHandler} are provided.
|
||||
* </ul>
|
||||
* <p>
|
||||
* An inheritor may consider to call this method for population common properties and perform
|
||||
* custom message serialization logic for the parameter #6.
|
||||
* Any custom data structure population can be achieved with full overriding of this method.
|
||||
* @param preparedStatement the {@link PreparedStatement} to populate columns based on the provided arguments
|
||||
* @param requestMessage the {@link Message} to store
|
||||
* @param groupId the group id for the message to store
|
||||
* @param region the region in the target table to distinguish different data base clients
|
||||
* @param priorityEnabled the flag to indicate if priority has to be stored
|
||||
* @throws SQLException the exception throws during data population
|
||||
*/
|
||||
public void setValues(PreparedStatement preparedStatement, Message<?> requestMessage, Object groupId, String region,
|
||||
boolean priorityEnabled) throws SQLException {
|
||||
|
||||
String groupKey = Objects.toString(UUIDConverter.getUUID(groupId), null);
|
||||
long createdDate = System.currentTimeMillis();
|
||||
String messageId = Objects.toString(UUIDConverter.getUUID(requestMessage.getHeaders().getId()), null);
|
||||
|
||||
preparedStatement.setString(1, messageId);
|
||||
preparedStatement.setString(2, groupKey);
|
||||
preparedStatement.setString(3, region);
|
||||
preparedStatement.setLong(4, createdDate);
|
||||
|
||||
Integer priority = requestMessage.getHeaders().get(IntegrationMessageHeaderAccessor.PRIORITY, Integer.class);
|
||||
|
||||
if (priorityEnabled && priority != null) {
|
||||
preparedStatement.setInt(5, priority);
|
||||
}
|
||||
else {
|
||||
preparedStatement.setNull(5, Types.NUMERIC);
|
||||
}
|
||||
|
||||
if (this.serializer != null) {
|
||||
byte[] messageBytes = this.serializer.convert(requestMessage);
|
||||
this.lobHandler.getLobCreator().setBlobAsBytes(preparedStatement, 6, messageBytes);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,88 +0,0 @@
|
||||
/*
|
||||
* Copyright 2002-2017 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.integration.jdbc.store.channel;
|
||||
|
||||
import java.sql.PreparedStatement;
|
||||
import java.sql.SQLException;
|
||||
import java.sql.Types;
|
||||
|
||||
import org.springframework.core.serializer.support.SerializingConverter;
|
||||
import org.springframework.integration.IntegrationMessageHeaderAccessor;
|
||||
import org.springframework.integration.util.UUIDConverter;
|
||||
import org.springframework.jdbc.support.lob.LobHandler;
|
||||
import org.springframework.messaging.Message;
|
||||
|
||||
/**
|
||||
* Callback to be used with {@link org.springframework.integration.jdbc.store.JdbcChannelMessageStore}.
|
||||
* <p>
|
||||
* Behavior is same as standard {@link org.springframework.jdbc.core.PreparedStatementSetter}, it takes in
|
||||
* extra {@code Message<?> requestMessage}, {@code Object groupId},
|
||||
* {@code String region} and {@code boolean priorityEnabled} as parameters used
|
||||
* for {@code addMessageToGroup} method in {@link org.springframework.integration.jdbc.store.JdbcChannelMessageStore}.
|
||||
*
|
||||
* @author Meherzad Lahewala
|
||||
* @since 5.0
|
||||
* @see org.springframework.jdbc.core.PreparedStatementSetter
|
||||
*/
|
||||
public class MessageGroupPreparedStatementSetter {
|
||||
|
||||
private final SerializingConverter serializer;
|
||||
|
||||
private final LobHandler lobHandler;
|
||||
|
||||
public MessageGroupPreparedStatementSetter(SerializingConverter serializer, LobHandler lobHandler) {
|
||||
this.serializer = serializer;
|
||||
this.lobHandler = lobHandler;
|
||||
}
|
||||
|
||||
protected MessageGroupPreparedStatementSetter() {
|
||||
this.serializer = null;
|
||||
this.lobHandler = null;
|
||||
}
|
||||
|
||||
public void setValues(PreparedStatement preparedStatement, Message<?> requestMessage, Object groupId, String region,
|
||||
boolean priorityEnabled) throws SQLException {
|
||||
String groupKey = getKey(groupId);
|
||||
long createdDate = System.currentTimeMillis();
|
||||
String messageId = getKey(requestMessage.getHeaders().getId());
|
||||
byte[] messageBytes = null;
|
||||
if (this.serializer != null) {
|
||||
messageBytes = this.serializer.convert(requestMessage);
|
||||
}
|
||||
|
||||
preparedStatement.setString(1, messageId);
|
||||
preparedStatement.setString(2, groupKey);
|
||||
preparedStatement.setString(3, region);
|
||||
preparedStatement.setLong(4, createdDate);
|
||||
|
||||
Integer priority = requestMessage.getHeaders().get(IntegrationMessageHeaderAccessor.PRIORITY, Integer.class);
|
||||
|
||||
if (priorityEnabled && priority != null) {
|
||||
preparedStatement.setInt(5, priority);
|
||||
}
|
||||
else {
|
||||
preparedStatement.setNull(5, Types.NUMERIC);
|
||||
}
|
||||
if (this.lobHandler != null) {
|
||||
this.lobHandler.getLobCreator().setBlobAsBytes(preparedStatement, 6, messageBytes);
|
||||
}
|
||||
}
|
||||
|
||||
protected String getKey(Object input) {
|
||||
return input == null ? null : UUIDConverter.getUUID(input).toString();
|
||||
}
|
||||
}
|
||||
@@ -37,7 +37,7 @@ import org.springframework.jdbc.support.lob.DefaultLobHandler;
|
||||
import org.springframework.jdbc.support.lob.LobHandler;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.test.annotation.DirtiesContext;
|
||||
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
|
||||
import org.springframework.test.context.junit4.SpringRunner;
|
||||
import org.springframework.transaction.PlatformTransactionManager;
|
||||
import org.springframework.transaction.TransactionDefinition;
|
||||
import org.springframework.transaction.TransactionStatus;
|
||||
@@ -49,10 +49,11 @@ import org.springframework.transaction.support.TransactionTemplate;
|
||||
* @author Gunnar Hillert
|
||||
* @author Gary Russell
|
||||
* @author Meherzad Lahewala
|
||||
* @author Artem Bilan
|
||||
*/
|
||||
|
||||
@RunWith(SpringJUnit4ClassRunner.class)
|
||||
@DirtiesContext // close at the end after class
|
||||
@RunWith(SpringRunner.class)
|
||||
@DirtiesContext
|
||||
public abstract class AbstractJdbcChannelMessageStoreTests {
|
||||
|
||||
protected static final String TEST_MESSAGE_GROUP = "AbstractJdbcChannelMessageStoreTests";
|
||||
@@ -97,6 +98,7 @@ public abstract class AbstractJdbcChannelMessageStoreTests {
|
||||
transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
|
||||
|
||||
transactionTemplate.execute(new TransactionCallbackWithoutResult() {
|
||||
|
||||
@Override
|
||||
protected void doInTransactionWithoutResult(TransactionStatus status) {
|
||||
messageStore.addMessageToGroup(TEST_MESSAGE_GROUP, message);
|
||||
@@ -111,7 +113,7 @@ public abstract class AbstractJdbcChannelMessageStoreTests {
|
||||
|
||||
@Test
|
||||
public void testAddAndGetCustomStatementSetter() {
|
||||
messageStore.setMessageGroupPreparedStatementSetter(getMessageGroupPreparedStatementSetter());
|
||||
messageStore.setPreparedStatementSetter(getMessageGroupPreparedStatementSetter());
|
||||
final Message<String> message = MessageBuilder.withPayload("Cartman and Kenny").build();
|
||||
|
||||
final TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
|
||||
@@ -120,6 +122,7 @@ public abstract class AbstractJdbcChannelMessageStoreTests {
|
||||
transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
|
||||
|
||||
transactionTemplate.execute(new TransactionCallbackWithoutResult() {
|
||||
|
||||
@Override
|
||||
protected void doInTransactionWithoutResult(TransactionStatus status) {
|
||||
messageStore.addMessageToGroup(TEST_MESSAGE_GROUP, message);
|
||||
@@ -130,19 +133,22 @@ public abstract class AbstractJdbcChannelMessageStoreTests {
|
||||
assertEquals(message.getHeaders().getId(), messageFromDb.getHeaders().getId());
|
||||
}
|
||||
|
||||
private MessageGroupPreparedStatementSetter getMessageGroupPreparedStatementSetter() {
|
||||
return new MessageGroupPreparedStatementSetter() {
|
||||
private ChannelMessageStorePreparedStatementSetter getMessageGroupPreparedStatementSetter() {
|
||||
return new ChannelMessageStorePreparedStatementSetter() {
|
||||
|
||||
private SerializingConverter serializer = new SerializingConverter();
|
||||
|
||||
private LobHandler lobHandler = new DefaultLobHandler();
|
||||
|
||||
@Override
|
||||
public void setValues(PreparedStatement preparedStatement, Message<?> requestMessage, Object groupId,
|
||||
String region, boolean priorityEnabled) throws SQLException {
|
||||
super.setValues(preparedStatement, requestMessage, groupId, region, priorityEnabled);
|
||||
byte[] messageBytes = serializer.convert(requestMessage);
|
||||
lobHandler.getLobCreator().setBlobAsBytes(preparedStatement, 6, messageBytes);
|
||||
byte[] messageBytes = this.serializer.convert(requestMessage);
|
||||
this.lobHandler.getLobCreator().setBlobAsBytes(preparedStatement, 6, messageBytes);
|
||||
}
|
||||
|
||||
};
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -365,35 +365,29 @@ If your database is not listed, you can easily extend the `AbstractChannelMessag
|
||||
|
||||
Since _version 4.0_, the `MESSAGE_SEQUENCE` column has been added to the table to ensure first-in-first-out (FIFO) queueing even when messages are stored in the same millisecond.
|
||||
|
||||
Since_version 5.0, by overloading `MessageGroupPreparedStatementSetter` class you can provide custom
|
||||
implementation to `PreparedStatementSetter` to support different columns or table structure
|
||||
or serialization strategy.
|
||||
For example, instead of default serialization to byte[], we can store its structure in JSON string.
|
||||
Since _version 5.0_, by overloading `ChannelMessageStorePreparedStatementSetter` class you can provide custom implementation for message insertion in the `JdbcChannelMessageStore`.
|
||||
It might be different columns or table structure or serialization strategy.
|
||||
For example, instead of default serialization to `byte[]`, we can store its structure in JSON string.
|
||||
|
||||
Below example uses the default implementation of `setValues` to store common columns and overrides the behavior
|
||||
just to store the message payload as varchar.
|
||||
Below example uses the default implementation of `setValues` to store common columns and overrides the behavior just to store the message payload as varchar.
|
||||
|
||||
[source,java]
|
||||
----
|
||||
public class JsonPreparedStatementSetter extends ChannelMessageStorePreparedStatementSetter {
|
||||
|
||||
public class CustomMessageGroupPreparedStatementSetter
|
||||
extends MessageGroupPreparedStatementSetter {
|
||||
public JsonPreparedStatementSetter() {
|
||||
super();
|
||||
}
|
||||
|
||||
public CustomMessageGroupPreparedStatementSetter() {
|
||||
super();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setValues(PreparedStatement preparedStatement, Message<?> requestMessage,
|
||||
Object groupId, String region, boolean priorityEnabled) throws SQLException {
|
||||
// Populate common columns
|
||||
super.setValues(preparedStatement, requestMessage, groupId, region, priorityEnabled);
|
||||
// Store message payload as varchar
|
||||
preparedStatement.setString(6, requestMessage.getPayload().toString());
|
||||
}
|
||||
@Override
|
||||
public void setValues(PreparedStatement preparedStatement, Message<?> requestMessage,
|
||||
Object groupId, String region, boolean priorityEnabled) throws SQLException {
|
||||
// Populate common columns
|
||||
super.setValues(preparedStatement, requestMessage, groupId, region, priorityEnabled);
|
||||
// Store message payload as varchar
|
||||
preparedStatement.setString(6, requestMessage.getPayload().toString());
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
----
|
||||
|
||||
[IMPORTANT]
|
||||
|
||||
@@ -294,7 +294,6 @@ See <<gemfire>> for more information.
|
||||
|
||||
==== Jdbc Message Channel Store Changes
|
||||
|
||||
The `JdbcMessageChannelStore` now provides setter to set `MessageGroupPreparedStatementSetter`,
|
||||
allowing users to easily customize `PreparedStatementSetter` in the store.
|
||||
The `JdbcMessageChannelStore` now provides setter for the `ChannelMessageStorePreparedStatementSetter` allowing users to customize a message insertion in the store.
|
||||
|
||||
See <<jdbc-message-store-channels>> for more information.
|
||||
|
||||
Reference in New Issue
Block a user