From d7d56a04445a04e0b8e9757e8bdb2dc7ee4bce4a Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 23 Oct 2017 11:41:04 -0400 Subject: [PATCH] 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 --- .../jdbc/store/JdbcChannelMessageStore.java | 132 ++++++----------- ...elMessageStorePreparedStatementSetter.java | 135 ++++++++++++++++++ .../MessageGroupPreparedStatementSetter.java | 88 ------------ .../AbstractJdbcChannelMessageStoreTests.java | 22 +-- src/reference/asciidoc/jdbc.adoc | 38 +++-- src/reference/asciidoc/whats-new.adoc | 3 +- 6 files changed, 211 insertions(+), 207 deletions(-) create mode 100644 spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/ChannelMessageStorePreparedStatementSetter.java delete mode 100644 spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/MessageGroupPreparedStatementSetter.java diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java index 01a78ad43c..3c50e0e28a 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java @@ -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 idCache = new HashSet(); - - 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 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 queryCache = new HashMap(); + 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 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 1 * and with {@link JdbcTemplate#setMaxRows(int)} set to 1. - * * @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 1 * and with {@link JdbcTemplate#setMaxRows(int)} set to 1. - * * @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> 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 {@link #DEFAULT_REGION}. - * * @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 *

Consider using this property when polling the database transactionally * using multiple parallel threads, meaning when the configured poller is configured * using a task executor.

- * *

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.

- * *

If using the provided {@link OracleChannelMessageStoreQueryProvider}, don't set {@link #usingIdCache} * to true, as the Oracle query will ignore locked rows.

- * *

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.

- * *

For this to work, you must setup the corresponding * {@link TransactionSynchronizationFactory}:

- * * 
 	 * {@code
 	 * 
@@ -348,10 +323,8 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
 	 * 
 	 * }
 	 *
- * * This {@link TransactionSynchronizationFactory} is then referenced in the * transaction configuration of the poller: - * * 
 	 * {@code
 	 * 
 	 * }
 	 *
- * * @param usingIdCache When true 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.

*

Only applicable if {@link #setUsingIdCache(boolean)} is set to * true

. - * * @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 diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/ChannelMessageStorePreparedStatementSetter.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/ChannelMessageStorePreparedStatementSetter.java new file mode 100644 index 0000000000..8bf4d7c160 --- /dev/null +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/ChannelMessageStorePreparedStatementSetter.java @@ -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}. + *

+ * 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}. + *

+ * 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): + *

+ *

    + *
  • 1 - messageId + *
  • 2 - groupKey + *
  • 3 - region + *
  • 4 - createdDate + *
  • 5 - priority if enabled, otherwise null + *
  • 6 - serialized message if {@link #serializer} and {@link #lobHandler} are provided. + *
+ *

+ * 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); + } + } + +} diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/MessageGroupPreparedStatementSetter.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/MessageGroupPreparedStatementSetter.java deleted file mode 100644 index 6b6d670cfb..0000000000 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/MessageGroupPreparedStatementSetter.java +++ /dev/null @@ -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}. - *

- * 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(); - } -} diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractJdbcChannelMessageStoreTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractJdbcChannelMessageStoreTests.java index 42d8439fbc..76aee8b5a3 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractJdbcChannelMessageStoreTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractJdbcChannelMessageStoreTests.java @@ -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 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); } + }; } + } diff --git a/src/reference/asciidoc/jdbc.adoc b/src/reference/asciidoc/jdbc.adoc index 0689bc00f4..5c4226272a 100644 --- a/src/reference/asciidoc/jdbc.adoc +++ b/src/reference/asciidoc/jdbc.adoc @@ -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] diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index c1fe7ceafa..36f7efc554 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -294,7 +294,6 @@ See <> 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 <> for more information.