From 34d0859515d68ab213b7e97ac4d526e4e374efe5 Mon Sep 17 00:00:00 2001 From: Meherzad Lahewala Date: Sun, 15 Oct 2017 17:05:23 +0530 Subject: [PATCH] INT-4081: Custom PreparedStSetter for JdbcMChStore JIRA: https://jira.spring.io/browse/INT-4081 * Add `MessageGroupPreparedStatementSetter` to let end-user to override the message insertion logic * Add tests and Docs --- .../jdbc/store/JdbcChannelMessageStore.java | 75 +++++++--------- .../MessageGroupPreparedStatementSetter.java | 88 +++++++++++++++++++ .../AbstractJdbcChannelMessageStoreTests.java | 49 ++++++++++- src/reference/asciidoc/jdbc.adoc | 31 +++++++ src/reference/asciidoc/whats-new.adoc | 7 ++ 5 files changed, 207 insertions(+), 43 deletions(-) create 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 55ba1e4e7e..01a78ad43c 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 @@ -40,8 +40,8 @@ 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.IntegrationMessageHeaderAccessor; 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; @@ -87,6 +87,7 @@ import org.springframework.util.StringUtils; * @author Gunnar Hillert * @author Artem Bilan * @author Gary Russell + * @author Meherzad Lahewala * @since 2.2 */ @ManagedResource @@ -145,6 +146,8 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto private volatile MessageRowMapper messageRowMapper; + private volatile MessageGroupPreparedStatementSetter messageGroupPreparedStatementSetter; + private volatile Map queryCache = new HashMap(); private volatile MessageGroupFactory messageGroupFactory = new SimpleMessageGroupFactory(); @@ -153,6 +156,7 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto private boolean priorityEnabled; + @SuppressWarnings("unused") private BeanFactory beanFactory; /** @@ -254,6 +258,20 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto this.messageRowMapper = messageRowMapper; } + /** + * 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 + * @since 5.0 + */ + public void setMessageGroupPreparedStatementSetter( + MessageGroupPreparedStatementSetter messageGroupPreparedStatementSetter) { + Assert.notNull(messageGroupPreparedStatementSetter, + "The provided MessageGroupPreparedStatementSetter must not be null."); + this.messageGroupPreparedStatementSetter = messageGroupPreparedStatementSetter; + } + /** * Sets the database specific {@link ChannelMessageStoreQueryProvider} to use. * The {@link JdbcChannelMessageStore} provides the SQL queries to retrieve messages from @@ -382,10 +400,12 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto /** * Check mandatory properties ({@link DataSource} and - * {@link #setChannelMessageStoreQueryProvider(ChannelMessageStoreQueryProvider)}). If no {@link MessageRowMapper} was - * explicitly set using {@link #setMessageRowMapper(MessageRowMapper)}, the default - * {@link MessageRowMapper} will be instantiate using the specified {@link #deserializer} - * and {@link #lobHandler}. + * {@link #setChannelMessageStoreQueryProvider(ChannelMessageStoreQueryProvider)}). If no {@link MessageRowMapper} + * and {@link MessageGroupPreparedStatementSetter} was explicitly set using + * {@link #setMessageRowMapper(MessageRowMapper)} and + * {@link #setMessageGroupPreparedStatementSetter(MessageGroupPreparedStatementSetter)} respectively, the default + * {@link MessageRowMapper} and {@link MessageGroupPreparedStatementSetter} 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} @@ -407,6 +427,10 @@ 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, + this.lobHandler); + } this.jdbcTemplate.afterPropertiesSet(); } @@ -414,55 +438,24 @@ 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. + * 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 */ @Override public MessageGroup addMessageToGroup(Object groupId, final Message message) { - - String groupKey = getKey(groupId); - - long createdDate = System.currentTimeMillis(); - - String messageId = getKey(message.getHeaders().getId()); - - byte[] messageBytes = this.serializer.convert(message); - - if (logger.isDebugEnabled()) { - logger.debug("Inserting message with id key=" + messageId); - } - try { this.jdbcTemplate.update(getQuery(this.channelMessageStoreQueryProvider.getCreateMessageQuery()), - ps -> { - ps.setString(1, messageId); - ps.setString(2, groupKey); - ps.setString(3, this.region); - ps.setLong(4, createdDate); - - Integer priority = message.getHeaders() - .get(IntegrationMessageHeaderAccessor.PRIORITY, Integer.class); - - if (JdbcChannelMessageStore.this.priorityEnabled && priority != null) { - ps.setInt(5, priority); - } - else { - ps.setNull(5, Types.NUMERIC); - } - - this.lobHandler.getLobCreator().setBlobAsBytes(ps, 6, messageBytes); - }); + ps -> this.messageGroupPreparedStatementSetter.setValues(ps, message, groupId, this.region, + this.priorityEnabled)); } catch (DuplicateKeyException e) { if (logger.isDebugEnabled()) { - logger.debug("The Message with id [" + messageId + "] already exists.\n" + - "Ignoring INSERT..."); + String messageId = getKey(message.getHeaders().getId()); + logger.debug("The Message with id [" + messageId + "] already exists.\nIgnoring INSERT..."); } } - return getMessageGroup(groupId); } 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 new file mode 100644 index 0000000000..6b6d670cfb --- /dev/null +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/MessageGroupPreparedStatementSetter.java @@ -0,0 +1,88 @@ +/* + * 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 3c237bbb2f..42d8439fbc 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * 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. @@ -20,6 +20,9 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; +import java.sql.PreparedStatement; +import java.sql.SQLException; + import javax.sql.DataSource; import org.junit.Before; @@ -27,8 +30,11 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.core.serializer.support.SerializingConverter; import org.springframework.integration.jdbc.store.JdbcChannelMessageStore; import org.springframework.integration.support.MessageBuilder; +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; @@ -42,6 +48,7 @@ import org.springframework.transaction.support.TransactionTemplate; /** * @author Gunnar Hillert * @author Gary Russell + * @author Meherzad Lahewala */ @RunWith(SpringJUnit4ClassRunner.class) @@ -50,6 +57,8 @@ public abstract class AbstractJdbcChannelMessageStoreTests { protected static final String TEST_MESSAGE_GROUP = "AbstractJdbcChannelMessageStoreTests"; + private static final String REGION = "AbstractJdbcChannelMessageStoreTests"; + @Autowired protected DataSource dataSource; @@ -64,7 +73,7 @@ public abstract class AbstractJdbcChannelMessageStoreTests { @Before public void init() throws Exception { messageStore = new JdbcChannelMessageStore(dataSource); - messageStore.setRegion("AbstractJdbcChannelMessageStoreTests"); + messageStore.setRegion(REGION); messageStore.setChannelMessageStoreQueryProvider(queryProvider); messageStore.afterPropertiesSet(); messageStore.removeMessageGroup("AbstractJdbcChannelMessageStoreTests"); @@ -100,4 +109,40 @@ public abstract class AbstractJdbcChannelMessageStoreTests { assertEquals(message.getHeaders().getId(), messageFromDb.getHeaders().getId()); } + @Test + public void testAddAndGetCustomStatementSetter() { + messageStore.setMessageGroupPreparedStatementSetter(getMessageGroupPreparedStatementSetter()); + final Message message = MessageBuilder.withPayload("Cartman and Kenny").build(); + + final TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager); + + transactionTemplate.setIsolationLevel(Isolation.READ_COMMITTED.value()); + transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED); + + transactionTemplate.execute(new TransactionCallbackWithoutResult() { + @Override + protected void doInTransactionWithoutResult(TransactionStatus status) { + messageStore.addMessageToGroup(TEST_MESSAGE_GROUP, message); + } + }); + Message messageFromDb = messageStore.pollMessageFromGroup(TEST_MESSAGE_GROUP); + assertNotNull(messageFromDb); + assertEquals(message.getHeaders().getId(), messageFromDb.getHeaders().getId()); + } + + private MessageGroupPreparedStatementSetter getMessageGroupPreparedStatementSetter() { + return new MessageGroupPreparedStatementSetter() { + + 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); + } + }; + } } diff --git a/src/reference/asciidoc/jdbc.adoc b/src/reference/asciidoc/jdbc.adoc index 0d43dfd525..0689bc00f4 100644 --- a/src/reference/asciidoc/jdbc.adoc +++ b/src/reference/asciidoc/jdbc.adoc @@ -365,6 +365,37 @@ 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. + +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 CustomMessageGroupPreparedStatementSetter + extends MessageGroupPreparedStatementSetter { + + 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()); + } +} + + +---- + [IMPORTANT] ===== Generally it is not recommended to use a relational database for the purpose of queuing. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 94e4175a72..c1fe7ceafa 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -291,3 +291,10 @@ See <> for more information. The `GemfireMetadataStore` now implements `ListenableMetadataStore`, allowing users to listen to cache events by providing `MetadataStoreListener` instances to the store. 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. + +See <> for more information.