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

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<String, String> queryCache = new HashMap<String, String>();
 
 	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);
 	}
 

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

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<String> 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);
+			}
+		};
+	}
 }

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.

src/reference/asciidoc/whats-new.adoc

@@ -291,3 +291,10 @@ See <<ip>> for more information.
 The `GemfireMetadataStore` now implements `ListenableMetadataStore`, allowing users to listen to cache events by providing `MetadataStoreListener` instances to the store.
 
 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.
+
+See <<jdbc-message-store-channels>> for more information.