INT-2674 - Add JdbcChannelMessageStore

INT-2674 - Added slow/failing JDBC Message Store Tests INT-2674 - Code Review Changes INT-2674 - Add support for HSQL INT-2674 - Add documentation * Refactor Test Cases * Code Review Changes INT-2674 - Code Review Changes INT-2674 - Code Review Changes INT-2674 - Renamed classes INT-2674 - Code Review Changes
2012-11-06 13:16:47 -05:00
parent 5b0b0d731f
commit a5d7de848d
54 changed files with 2168 additions and 39 deletions
--- a/build.gradle
+++ b/build.gradle
@@ -439,10 +439,18 @@ project('spring-integration-jdbc') {
 		testCompile "com.h2database:h2:1.3.160"
 		testCompile "hsqldb:hsqldb:1.8.0.10"
 		testCompile "org.apache.derby:derby:10.5.3.0_1"
+		testCompile "org.apache.derby:derbyclient:10.5.3.0_1"
+
 		testCompile "org.aspectj:aspectjrt:$aspectjVersion"
 		testCompile "org.aspectj:aspectjweaver:$aspectjVersion"
 		testCompile "org.powermock:powermock-module-junit4:1.4.12"
 		testCompile "org.powermock:powermock-api-mockito:1.4.12"
+
+		testCompile "postgresql:postgresql:9.1-901-1.jdbc4"
+		testCompile "mysql:mysql-connector-java:5.1.21"
+		testCompile "commons-dbcp:commons-dbcp:1.4"
+		//testCompile "com.oracle:ojdbc6:11.2.0.3"
+
 	}
 	bundlor {
 		bundleSymbolicName = 'org.springframework.integration.jdbc'
--- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java
+++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java
@@ -38,7 +38,9 @@ import org.springframework.core.serializer.support.DeserializingConverter;
 import org.springframework.core.serializer.support.SerializingConverter;
 import org.springframework.dao.DuplicateKeyException;
 import org.springframework.integration.Message;
+import org.springframework.integration.MessageChannel;
 import org.springframework.integration.MessageHeaders;
+import org.springframework.integration.jdbc.store.JdbcChannelMessageStore;
 import org.springframework.integration.store.AbstractMessageGroupStore;
 import org.springframework.integration.store.MessageGroup;
 import org.springframework.integration.store.MessageStore;
@@ -63,9 +65,17 @@ import org.springframework.util.StringUtils;
 * tables are packaged as <code>org/springframework/integration/jdbc/schema-*.sql</code>, where <code>*</code> is the
 * target database type.
 *
+ * Notice: Starting with Spring Integration 3.0, this class will move to package:
+ * <code>org.springframework.integration.jdbc.store</code>.
+ *
+ * If you intend backing a {@link MessageChannel} using a JDBC-based Message Store,
+ * please consider using the channel-specific {@link JdbcChannelMessageStore} instead.
+ *
 * @author Dave Syer
 * @author Oleg Zhurakousky
 * @author Matt Stine
+ * @author Gunnar Hillert
+ *
 * @since 2.0
 */
@ManagedResource
--- 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
@@ -0,0 +1,647 @@
+/*
+ * Copyright 2002-2012 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;
+
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Types;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+
+import javax.sql.DataSource;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.core.serializer.Deserializer;
+import org.springframework.core.serializer.Serializer;
+import org.springframework.core.serializer.support.DeserializingConverter;
+import org.springframework.core.serializer.support.SerializingConverter;
+import org.springframework.integration.Message;
+import org.springframework.integration.MessageHeaders;
+import org.springframework.integration.jdbc.JdbcMessageStore;
+import org.springframework.integration.jdbc.store.channel.DerbyChannelMessageStoreQueryProvider;
+import org.springframework.integration.jdbc.store.channel.MessageRowMapper;
+import org.springframework.integration.jdbc.store.channel.MySqlChannelMessageStoreQueryProvider;
+import org.springframework.integration.jdbc.store.channel.OracleChannelMessageStoreQueryProvider;
+import org.springframework.integration.jdbc.store.channel.PostgresChannelMessageStoreQueryProvider;
+import org.springframework.integration.jdbc.store.channel.ChannelMessageStoreQueryProvider;
+import org.springframework.integration.store.AbstractMessageGroupStore;
+import org.springframework.integration.store.MessageGroup;
+import org.springframework.integration.store.MessageGroupStore;
+import org.springframework.integration.store.MessageStore;
+import org.springframework.integration.store.SimpleMessageGroup;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.transaction.TransactionSynchronizationFactory;
+import org.springframework.integration.util.UUIDConverter;
+import org.springframework.jdbc.core.JdbcOperations;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.PreparedStatementSetter;
+import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
+import org.springframework.jdbc.support.lob.DefaultLobHandler;
+import org.springframework.jdbc.support.lob.LobHandler;
+import org.springframework.jmx.export.annotation.ManagedAttribute;
+import org.springframework.jmx.export.annotation.ManagedMetric;
+import org.springframework.jmx.export.annotation.ManagedResource;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * <p>
+ * Channel-specific implementation of {@link MessageGroupStore} using a relational
+ * database via JDBC.
+ *
+ * This message store shall be used for message channels only.
+ * </p>
+ * <p>
+ * <strong>NOTICE</strong>: This implementation may change for Spring Integration
+ * 3.0. It is provided for use-cases where the current {@link JdbcMessageStore}
+ * is not delivering the desired performance characteristics.
+ * </p>
+ *
+ * <p>
+ * As such, the {@link JdbcChannelMessageStore} uses database specific SQL queries.
+ * </p>
+ * <p>
+ * Contrary to the {@link JdbcMessageStore}, this implementation uses one single
+ * database table only. The SQL scripts to create the necessary table are packaged
+ * under <code>org/springframework/integration/jdbc/messagestore/channel/schema-*.sql</code>,
+ * where <code>*</code> denotes the target database type.
+ * </p
+ * >
+ * @author Gunnar Hillert
+ * @since 2.2
+ */
+@ManagedResource
+public class JdbcChannelMessageStore extends AbstractMessageGroupStore implements InitializingBean {
+
+	private static final Log logger = LogFactory.getLog(JdbcChannelMessageStore.class);
+	private final Set<String> idCache = Collections.newSetFromMap(new ConcurrentHashMap<String,Boolean>());
+
+	/**
+	 * 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
+	 * may use the same message store by providing a distinct region name.
+	 */
+	public static final String DEFAULT_REGION = "DEFAULT";
+
+	private ChannelMessageStoreQueryProvider channelMessageStoreQueryProvider;
+
+	public static final int DEFAULT_LONG_STRING_LENGTH = 2500;
+
+	/**
+	 * The name of the message header that stores a flag to indicate that the message has been saved. This is an
+	 * optimization for the put method.
+	 */
+	public static final String SAVED_KEY = JdbcChannelMessageStore.class.getSimpleName() + ".SAVED";
+
+	/**
+	 * The name of the message header that stores a timestamp for the time the message was inserted.
+	 */
+	public static final String CREATED_DATE_KEY = JdbcChannelMessageStore.class.getSimpleName() + ".CREATED_DATE";
+
+	private volatile String region = DEFAULT_REGION;
+
+	private volatile String tablePrefix = DEFAULT_TABLE_PREFIX;
+
+	private volatile JdbcTemplate jdbcTemplate;
+
+	private volatile DeserializingConverter deserializer;
+
+	private volatile SerializingConverter serializer;
+
+	private volatile LobHandler lobHandler = new DefaultLobHandler();
+
+	private volatile MessageRowMapper messageRowMapper;
+
+	private volatile Map<String, String> queryCache = new HashMap<String, String>();
+
+	private boolean usingIdCache = false;
+
+	/**
+	 * Convenient constructor for configuration use.
+	 */
+	public JdbcChannelMessageStore() {
+		deserializer = new DeserializingConverter();
+		serializer = new SerializingConverter();
+	}
+
+	/**
+	 * 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();
+		jdbcTemplate = new JdbcTemplate(dataSource);
+
+		this.jdbcTemplate.setFetchSize(1);
+		this.jdbcTemplate.setMaxRows(1);
+
+		this.jdbcTemplate.afterPropertiesSet();
+	}
+
+	/**
+	 * The JDBC {@link DataSource} to use when interacting with the database.
+	 * 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) {
+		jdbcTemplate = new JdbcTemplate(dataSource);
+
+		this.jdbcTemplate.setFetchSize(1);
+		this.jdbcTemplate.setMaxRows(1);
+
+		this.jdbcTemplate.afterPropertiesSet();
+	}
+
+	/**
+	 * A converter for deserializing byte arrays to messages.
+	 *
+	 * @param deserializer the deserializer to set
+	 */
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	public void setDeserializer(Deserializer<? extends Message<?>> deserializer) {
+		this.deserializer = new DeserializingConverter((Deserializer) deserializer);
+	}
+
+	/**
+	 * 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) {
+		Assert.notNull(jdbcTemplate, "The provided jdbcTemplate must not be null.");
+		this.jdbcTemplate = jdbcTemplate;
+	}
+
+	/**
+	 * Method not implemented.
+	 * @throws UnsupportedOperationException
+	 */
+	public void setLastReleasedSequenceNumberForGroup(Object groupId, final int sequenceNumber) {
+		throw new UnsupportedOperationException("Not implemented");
+	}
+
+	/**
+	 * 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) {
+		Assert.notNull(lobHandler, "The provided LobHandler must not be null.");
+		this.lobHandler = lobHandler;
+	}
+
+	/**
+	 * 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) {
+		Assert.notNull(messageRowMapper, "The provided MessageRowMapper must not be null.");
+		this.messageRowMapper = messageRowMapper;
+	}
+
+	/**
+	 * <p>
+	 * Sets the database specific {@link ChannelMessageStoreQueryProvider} to use. The {@link JdbcChannelMessageStore}
+	 * provides the SQL queries to retrieve messages from the database. The
+	 * following {@link ChannelMessageStoreQueryProvider} are provided:
+	 * </p>
+	 * <ul>
+	 *     <li>{@link DerbyChannelMessageStoreQueryProvider}</li>
+	 *     <li>{@link MySqlChannelMessageStoreQueryProvider}</li>
+	 *     <li>{@link OracleChannelMessageStoreQueryProvider}</li>
+	 *     <li>{@link PostgresChannelMessageStoreQueryProvider}</li>
+	 * </ul>
+	 * <p>
+	 * Beyond, you can provide your own query implementations, in case you need
+	 * to support additional databases and/or need to fine-tune the queries for
+	 * your requirements.
+	 * </p>
+	 *
+	 * @param channelMessageStoreQueryProvider Must not be null.
+	 */
+	public void setChannelMessageStoreQueryProvider(ChannelMessageStoreQueryProvider channelMessageStoreQueryProvider) {
+		Assert.notNull(channelMessageStoreQueryProvider, "The provided channelMessageStoreQueryProvider must not be null.");
+		this.channelMessageStoreQueryProvider = channelMessageStoreQueryProvider;
+	}
+
+	/**
+	 * 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) {
+		this.region = region;
+	}
+
+	/**
+	 * A converter for serializing messages to byte arrays for storage.
+	 *
+	 * @param serializer The serializer to set
+	 */
+	@SuppressWarnings("unchecked")
+	public void setSerializer(Serializer<? super Message<?>> serializer) {
+		Assert.notNull(serializer, "The provided serializer must not be null.");
+		this.serializer = new SerializingConverter((Serializer<Object>) serializer);
+	}
+
+	/**
+	 * 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) {
+		this.tablePrefix = tablePrefix;
+	}
+
+	/**
+	 * <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>
+	 * {@code
+	 * <int:transaction-synchronization-factory id="syncFactory">
+	 *     <int:after-commit   expression="@jdbcChannelMessageStore.removeFromIdCache(headers.id.toString())" />
+	 *     <int:after-rollback expression="@jdbcChannelMessageStore.removeFromIdCache(headers.id.toString())" />
+	 * </int:transaction-synchronization-factory>
+	 * }
+	 * </pre>
+	 *
+	 * This {@link TransactionSynchronizationFactory} is then referenced in the
+	 * transaction configuration of the poller:
+	 *
+	 * <pre>
+	 * {@code
+	 * <int:poller fixed-delay="300" receive-timeout="500"
+	 *     max-messages-per-poll="1" task-executor="pool">
+	 *     <int:transactional propagation="REQUIRED" synchronization-factory="syncFactory"
+	 *         isolation="READ_COMMITTED" transaction-manager="transactionManager" />
+	 * </int:poller>
+	 * }
+	 * </pre>
+	 *
+	 * @param usingIdCache When <code>true</code> the id cache will be used.
+	 */
+	public void setUsingIdCache(boolean usingIdCache) {
+		this.usingIdCache = usingIdCache;
+	}
+
+	/**
+	 * 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}.
+	 *
+	 * 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
+	 */
+	public void afterPropertiesSet() throws Exception {
+		Assert.state(jdbcTemplate != null, "A DataSource or JdbcTemplate must be provided");
+		Assert.notNull(this.channelMessageStoreQueryProvider, "A channelMessageStoreQueryProvider must be provided.");
+
+		if (this.messageRowMapper == null) {
+			this.messageRowMapper = new MessageRowMapper(this.deserializer, this.lobHandler);
+		}
+
+		if (this.jdbcTemplate.getFetchSize() != 1 && logger.isWarnEnabled()) {
+			logger.warn("The jdbcTemplate's fetchsize is not 1 but %s. This may cause FIFO issues with Oracle databases.");
+		}
+
+	}
+
+	/**
+	 * 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 groupdId (Channel
+	 * Identifier) is converted to a String-based UUID identifier.
+	 *
+	 * @param groupId the group id to store the message under
+	 * @param message a message
+	 */
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	public MessageGroup addMessageToGroup(Object groupId, Message<?> message) {
+
+		final String groupKey = getKey(groupId);
+
+		final long createdDate = System.currentTimeMillis();
+		final Message<?> result = MessageBuilder.fromMessage(message).setHeader(SAVED_KEY, Boolean.TRUE)
+				.setHeader(CREATED_DATE_KEY, new Long(createdDate)).build();
+
+		final Map innerMap = (Map) new DirectFieldAccessor(result.getHeaders()).getPropertyValue("headers");
+		// using reflection to set ID since it is immutable through MessageHeaders
+		innerMap.put(MessageHeaders.ID, message.getHeaders().get(MessageHeaders.ID));
+
+		final String messageId = getKey(result.getHeaders().getId());
+		final byte[] messageBytes = serializer.convert(result);
+
+		jdbcTemplate.update(getQuery(channelMessageStoreQueryProvider.getCreateMessageQuery()), new PreparedStatementSetter() {
+			public void setValues(PreparedStatement ps) throws SQLException {
+				if (logger.isDebugEnabled()){
+					logger.debug("Inserting message with id key=" + messageId);
+				}
+				ps.setString(1, messageId);
+				ps.setString(2, groupKey);
+				ps.setString(3, region);
+				ps.setLong(4, createdDate);
+				lobHandler.getLobCreator().setBlobAsBytes(ps, 5, messageBytes);
+			}
+		});
+
+		return getMessageGroup(groupId);
+	}
+
+	/**
+	 * Method not implemented.
+	 * @throws UnsupportedOperationException
+	 */
+	public void completeGroup(Object groupId) {
+		throw new UnsupportedOperationException("Not implemented");
+	}
+
+	/**
+	 * 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
+	 */
+	protected Message<?> doPollForMessage(String groupIdKey) {
+
+		final NamedParameterJdbcTemplate namedParameterJdbcTemplate = new NamedParameterJdbcTemplate(jdbcTemplate);
+		final MapSqlParameterSource parameters = new MapSqlParameterSource();
+
+		parameters.addValue("region", region);
+		parameters.addValue("group_key", groupIdKey);
+
+		final String query;
+
+		synchronized (idCache) {
+			if (this.usingIdCache && !this.idCache.isEmpty()) {
+				query = getQuery(this.channelMessageStoreQueryProvider.getPollFromGroupExcludeIdsQuery());
+				parameters.addValue("message_ids", idCache);
+			} else {
+				query = getQuery(this.channelMessageStoreQueryProvider.getPollFromGroupQuery());
+			}
+		}
+
+		final List<Message<?>> messages = namedParameterJdbcTemplate.query(query, parameters, messageRowMapper);
+
+		Assert.isTrue(messages.size() == 0 || messages.size() == 1);
+		if (messages.size() > 0){
+
+			final Message<?>message = messages.get(0);
+			final String messageId = message.getHeaders().getId().toString();
+
+			if (this.usingIdCache) {
+
+				boolean added = this.idCache.add(messageId);
+
+				if (logger.isDebugEnabled()) {
+					logger.debug(String.format("Polled message with id '%s' added: '%s'.", messageId, added));
+				}
+			}
+
+			return message;
+		}
+		return null;
+	}
+
+	/**
+	 * Helper method that converts the channel id to a UUID.
+	 *
+	 * @param input
+	 * @return
+	 */
+	private String getKey(Object input) {
+		return input == null ? null : UUIDConverter.getUUID(input).toString();
+	}
+
+	/**
+	 * Method not implemented.
+	 * @throws UnsupportedOperationException
+	 */
+	@ManagedAttribute
+	public long getMessageCount() {
+		throw new UnsupportedOperationException("Not implemented");
+	}
+
+	/**
+	 * Method not implemented.
+	 * @throws UnsupportedOperationException
+	 */
+	@Override
+	@ManagedAttribute
+	public int getMessageCountForAllMessageGroups() {
+		throw new UnsupportedOperationException("Not implemented");
+	}
+
+	/**
+	 * Not fully used. Only wraps the provided group id.
+	 */
+	public MessageGroup getMessageGroup(Object groupId) {
+		return new SimpleMessageGroup(groupId);
+	}
+
+	/**
+	 * Method not implemented.
+	 * @throws UnsupportedOperationException
+	 */
+	@Override
+	@ManagedAttribute
+	public int getMessageGroupCount() {
+		throw new UnsupportedOperationException("Not implemented");
+	}
+
+	/**
+	 * 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
+	 */
+	protected String getQuery(String sqlQuery) {
+		String query = queryCache.get(sqlQuery);
+
+		if (query == null) {
+			query = StringUtils.replace(sqlQuery, "%PREFIX%", tablePrefix);
+			queryCache.put(sqlQuery, query);
+		}
+
+		return query;
+	}
+
+	/**
+	 * Method not implemented.
+	 * @throws UnsupportedOperationException
+	 */
+	public Iterator<MessageGroup> iterator() {
+		throw new UnsupportedOperationException("Not implemented");
+	}
+
+	/**
+	 * Returns the number of messages persisted for the specified channel id (groupId)
+	 * and the specified region ({@link #setRegion(String)}).
+	 */
+	@ManagedAttribute
+	public int messageGroupSize(Object groupId) {
+		final String key = getKey(groupId);
+		return jdbcTemplate.queryForInt(getQuery(channelMessageStoreQueryProvider.getCountAllMessagesInGroupQuery()), key, this.region);
+	}
+
+	/**
+	 * Polls the database for a new message that is persisted for the given
+	 * group id which represents the channel identifier.
+	 */
+	public Message<?> pollMessageFromGroup(Object groupId) {
+
+		final String key = getKey(groupId);
+		final Message<?> polledMessage = this.doPollForMessage(key);
+
+		if (polledMessage != null){
+			this.removeMessageFromGroup(groupId, polledMessage);
+		}
+
+		return polledMessage;
+	}
+
+	/**
+	 * Remove a single message from the database.
+	 *
+	 * @param groupId The channel id to remove the message from
+	 * @param messageToRemove The message to remove
+	 *
+	 */
+	public MessageGroup removeMessageFromGroup(Object groupId, Message<?> messageToRemove) {
+
+		final UUID id = messageToRemove.getHeaders().getId();
+
+		int updated = jdbcTemplate.update(getQuery(channelMessageStoreQueryProvider.getDeleteMessageQuery()), new Object[] { getKey(id), getKey(groupId), region }, new int[] {
+					Types.VARCHAR, Types.VARCHAR, Types.VARCHAR });
+
+		if (updated != 0) {
+			logger.debug(String.format("Message with id '%s' was deleted.", id));
+		} else {
+			logger.warn(String.format("Message with id '%s' was not deleted.", id));
+		}
+
+		return getMessageGroup(groupId);
+	}
+
+	/**
+	 * <p>Remove a Message Id from the idCache. Should be used in conjunction
+	 * with the Spring Integration Transaction Synchronization feature to remove
+	 * a message from the Message Id cache once a transaction either succeeded or
+	 * rolled back.</p>
+	 * <p>Only applicable if {@link #setUsingIdCache(boolean)} is set to
+	 * <code>true</code></p>.
+	 *
+	 * @param messageId
+	 */
+	public void removeFromIdCache(String messageId) {
+		if (logger.isDebugEnabled()) {
+			logger.debug("Removing Message Id:" + messageId);
+		}
+		this.idCache.remove(messageId);
+	}
+
+	/**
+	 * 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
+	public int getSizeOfIdCache() {
+		return this.idCache.size();
+	}
+
+	/**
+	 * Will remove all messages from the message channel.
+	 */
+	public void removeMessageGroup(Object groupId) {
+
+		final String groupKey = getKey(groupId);
+
+		jdbcTemplate.update(getQuery(channelMessageStoreQueryProvider.getDeleteMessageGroupQuery()), new PreparedStatementSetter() {
+			public void setValues(PreparedStatement ps) throws SQLException {
+				if (logger.isDebugEnabled()){
+					logger.debug("Marking messages with group key=" + groupKey);
+				}
+				ps.setString(1, groupKey);
+				ps.setString(2, region);
+			}
+		});
+
+	}
+
+}
--- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/AbstractChannelMessageStoreQueryProvider.java
+++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/AbstractChannelMessageStoreQueryProvider.java
@@ -0,0 +1,50 @@
+/*
+ * Copyright 2002-2012 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;
+
+
+/**
+ * @author Gunnar Hillert
+ * @since 2.2
+ */
+public abstract class AbstractChannelMessageStoreQueryProvider implements ChannelMessageStoreQueryProvider {
+
+	public String getCountAllMessagesInGroupQuery() {
+		return "SELECT COUNT(MESSAGE_ID) from %PREFIX%CHANNEL_MESSAGE where GROUP_KEY=? and REGION=?";
+	}
+
+	public abstract String getPollFromGroupExcludeIdsQuery();
+	public abstract String getPollFromGroupQuery();
+
+	public String getMessageQuery() {
+		return "SELECT MESSAGE_ID, CREATED_DATE, MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE where MESSAGE_ID=? and GROUP_KEY=? and REGION=?";
+	}
+
+	public String getMessageCountForRegionQuery() {
+		return "SELECT COUNT(MESSAGE_ID) from %PREFIX%CHANNEL_MESSAGE where REGION=?";
+	}
+
+	public String getDeleteMessageQuery() {
+		return "DELETE from %PREFIX%CHANNEL_MESSAGE where MESSAGE_ID=? and GROUP_KEY=? and REGION=?";
+	}
+
+	public String getCreateMessageQuery() {
+		return "INSERT into %PREFIX%CHANNEL_MESSAGE(MESSAGE_ID, GROUP_KEY, REGION, CREATED_DATE, MESSAGE_BYTES)"
+				+ " values (?, ?, ?, ?, ?)";
+	}
+
+	public String getDeleteMessageGroupQuery() {
+		return "DELETE from %PREFIX%CHANNEL_MESSAGE where GROUP_KEY=? and REGION=?";
+	}
+
+}
--- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/ChannelMessageStoreQueryProvider.java
+++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/ChannelMessageStoreQueryProvider.java
@@ -0,0 +1,86 @@
+/*
+ * Copyright 2002-2012 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 org.springframework.integration.jdbc.store.JdbcChannelMessageStore;
+
+
+/**
+ * Common interface used in order to configure the
+ * {@link JdbcChannelMessageStore} to provide database-specific queries.
+ *
+ * @author Gunnar Hillert
+ * @since 2.2
+ */
+public interface ChannelMessageStoreQueryProvider {
+
+	/**
+	 * Get the query used to retrieve a count of all messages currently persisted
+	 * for a channel.
+	 *
+	 * @return Sql Query
+	 */
+	String getCountAllMessagesInGroupQuery();
+
+	/**
+	 * Get the query used to retrieve the oldest message for a channel excluding
+	 * messages that match the provided message ids.
+	 *
+	 * @return Sql Query
+	 */
+	String getPollFromGroupExcludeIdsQuery();
+
+	/**
+	 * Get the query used to retrieve the oldest message for a channel.
+	 *
+	 * @return Sql Query
+	 */
+	String getPollFromGroupQuery();
+
+	/**
+	 * Query that retrieves a message for the provided message id, channel and
+	 * region.
+	 *
+	 * @return Sql Query
+	 */
+	String getMessageQuery();
+
+	/**
+	 * Query that retrieve a count of all messages for a region.
+	 *
+	 * @return Sql Query
+	 */
+	String getMessageCountForRegionQuery();
+
+	/**
+	 * Query to delete a single message from the database.
+	 *
+	 * @return Sql Query
+	 */
+	String getDeleteMessageQuery();
+
+	/**
+	 * Query to add a single message to the database.
+	 *
+	 * @return Sql Query
+	 */
+	String getCreateMessageQuery();
+
+	/**
+	 * Query to delete all messages that belong to a specific channel.
+	 *
+	 * @return Sql Query
+	 */
+	String getDeleteMessageGroupQuery();
+
+}
--- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/DerbyChannelMessageStoreQueryProvider.java
+++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/DerbyChannelMessageStoreQueryProvider.java
@@ -0,0 +1,37 @@
+/*
+ * Copyright 2002-2012 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;
+
+/**
+ * @author Gunnar Hillert
+ * @since 2.2
+ *
+ * https://blogs.oracle.com/kah/entry/derby_10_5_preview_fetch
+ */
+public class DerbyChannelMessageStoreQueryProvider extends AbstractChannelMessageStoreQueryProvider {
+
+	@Override
+	public String getPollFromGroupExcludeIdsQuery() {
+		return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
+				"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
+				"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) order by CREATED_DATE ASC FETCH FIRST ROW ONLY";
+	}
+
+	@Override
+	public String getPollFromGroupQuery() {
+		return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
+				"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
+				"order by CREATED_DATE ASC FETCH FIRST ROW ONLY";
+	}
+
+}
--- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/HsqlChannelMessageStoreQueryProvider.java
+++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/HsqlChannelMessageStoreQueryProvider.java
@@ -0,0 +1,36 @@
+/*
+ * Copyright 2002-2012 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;
+
+/**
+ * @author Gunnar Hillert
+ * @since 2.2
+ *
+ */
+public class HsqlChannelMessageStoreQueryProvider extends AbstractChannelMessageStoreQueryProvider {
+
+	@Override
+	public String getPollFromGroupExcludeIdsQuery() {
+		return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
+				"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
+				"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) order by CREATED_DATE ASC LIMIT 1";
+	}
+
+	@Override
+	public String getPollFromGroupQuery() {
+		return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
+				"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
+				"order by CREATED_DATE ASC LIMIT 1";
+	}
+
+}
--- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/MessageRowMapper.java
+++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/MessageRowMapper.java
@@ -0,0 +1,46 @@
+/*
+ * Copyright 2002-2012 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.ResultSet;
+import java.sql.SQLException;
+
+import org.springframework.core.serializer.support.DeserializingConverter;
+import org.springframework.integration.Message;
+import org.springframework.jdbc.core.RowMapper;
+import org.springframework.jdbc.support.lob.LobHandler;
+
+/**
+ * Convenience class to be used to unpack a {@link Message} from a result set
+ * row. Uses column named in the result set to extract the required data, so
+ * that select clause ordering is unimportant.
+ *
+ * @author Gunnar Hillert
+ * @since 2.2
+ *
+ */
+public class MessageRowMapper implements RowMapper<Message<?>> {
+
+	private final DeserializingConverter deserializer;
+	private final LobHandler lobHandler;
+
+	public MessageRowMapper(DeserializingConverter deserializer, LobHandler lobHandler) {
+		this.deserializer = deserializer;
+		this.lobHandler = lobHandler;
+	}
+
+	public Message<?> mapRow(ResultSet rs, int rowNum) throws SQLException {
+		return (Message<?>) deserializer.convert(lobHandler.getBlobAsBytes(rs, "MESSAGE_BYTES"));
+	}
+
+}
--- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/MySqlChannelMessageStoreQueryProvider.java
+++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/MySqlChannelMessageStoreQueryProvider.java
@@ -0,0 +1,35 @@
+/*
+ * Copyright 2002-2012 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;
+
+/**
+ * @author Gunnar Hillert
+ * @since 2.2
+ */
+public class MySqlChannelMessageStoreQueryProvider extends AbstractChannelMessageStoreQueryProvider {
+
+	@Override
+	public String getPollFromGroupExcludeIdsQuery() {
+		return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
+				"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
+				"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) order by CREATED_DATE ASC LIMIT 1";
+	}
+
+	@Override
+	public String getPollFromGroupQuery() {
+		return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
+				"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
+				"order by CREATED_DATE ASC LIMIT 1";
+	}
+
+}
--- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/OracleChannelMessageStoreQueryProvider.java
+++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/OracleChannelMessageStoreQueryProvider.java
@@ -0,0 +1,44 @@
+/*
+ * Copyright 2002-2012 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 org.springframework.integration.jdbc.store.JdbcChannelMessageStore;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+/**
+ * Contains Oracle-specific queries for the {@link JdbcChannelMessageStore}.
+ * Please ensure that the used {@link JdbcTemplate}'s fetchSize property is <code>1</code>.
+ *
+ * Fore more details, please see: http://stackoverflow.com/questions/6117254/force-oracle-to-return-top-n-rows-with-skip-locked
+ *
+ * @author Gunnar Hillert
+ * @since 2.2
+ */
+public class OracleChannelMessageStoreQueryProvider extends AbstractChannelMessageStoreQueryProvider {
+
+	@Override
+	public String getPollFromGroupExcludeIdsQuery() {
+		return
+				"SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
+				"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
+				"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) order by CREATED_DATE ASC FOR UPDATE SKIP LOCKED";
+	}
+
+	@Override
+	public String getPollFromGroupQuery() {
+		return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
+				"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
+				"order by CREATED_DATE ASC FOR UPDATE SKIP LOCKED";
+	}
+
+}
--- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/PostgresChannelMessageStoreQueryProvider.java
+++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/PostgresChannelMessageStoreQueryProvider.java
@@ -0,0 +1,35 @@
+/*
+ * Copyright 2002-2012 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;
+
+/**
+ * @author Gunnar Hillert
+ * @since 2.2
+ */
+public class PostgresChannelMessageStoreQueryProvider extends AbstractChannelMessageStoreQueryProvider {
+
+	@Override
+	public String getPollFromGroupExcludeIdsQuery() {
+		return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
+				"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
+				"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) order by CREATED_DATE ASC LIMIT 1 FOR UPDATE";
+	}
+
+	@Override
+	public String getPollFromGroupQuery() {
+		return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
+				"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
+				"order by CREATED_DATE ASC LIMIT 1 FOR UPDATE";
+	}
+
+}
--- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/package-info.java
+++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/package-info.java
@@ -0,0 +1,4 @@
+/**
+ * Provides support classes for the JdbcChannelMessageStore.
+ */
+package org.springframework.integration.jdbc.store.channel;
--- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/package-info.java
+++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/package-info.java
@@ -0,0 +1,4 @@
+/**
+ * Provides JDBC-backed Message Store implementations.
+ */
+package org.springframework.integration.jdbc.store;
--- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-derby.sql
+++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-derby.sql
@@ -0,0 +1,10 @@
+CREATE TABLE INT_CHANNEL_MESSAGE (
+	MESSAGE_ID CHAR(36) NOT NULL,
+	GROUP_KEY CHAR(36) NOT NULL,
+	CREATED_DATE BIGINT NOT NULL,
+	MESSAGE_BYTES BLOB,
+	REGION VARCHAR(100) NOT NULL,
+	constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
+);
+
+CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE);
--- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-drop-derby.sql
+++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-drop-derby.sql
@@ -0,0 +1 @@
+DROP TABLE INT_CHANNEL_MESSAGE;
--- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-drop-hsql.sql
+++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-drop-hsql.sql
@@ -0,0 +1 @@
+DROP TABLE INT_CHANNEL_MESSAGE;
--- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-drop-mysql.sql
+++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-drop-mysql.sql
@@ -0,0 +1,2 @@
+DROP TABLE INT_CHANNEL_MESSAGE;
+
--- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-drop-oracle.sql
+++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-drop-oracle.sql
@@ -0,0 +1,3 @@
+DROP INDEX INT_CHANNEL_MSG_DATE_IDX;
+DROP TABLE INT_CHANNEL_MESSAGE;
+
--- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-drop-postgres.sql
+++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-drop-postgres.sql
@@ -0,0 +1,2 @@
+DROP TABLE INT_CHANNEL_MESSAGE;
+
--- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-hsql.sql
+++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-hsql.sql
@@ -0,0 +1,10 @@
+CREATE TABLE INT_CHANNEL_MESSAGE (
+	MESSAGE_ID CHAR(36) NOT NULL,
+	GROUP_KEY CHAR(36) NOT NULL,
+	CREATED_DATE BIGINT NOT NULL,
+	MESSAGE_BYTES LONGVARBINARY,
+	REGION VARCHAR(100) NOT NULL,
+	constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
+);
+
+CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE);
--- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-mysql.sql
+++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-mysql.sql
@@ -0,0 +1,11 @@
+CREATE TABLE INT_CHANNEL_MESSAGE (
+	MESSAGE_ID CHAR(36) NOT NULL,
+	GROUP_KEY CHAR(36) NOT NULL,
+	CREATED_DATE BIGINT NOT NULL,
+	MESSAGE_BYTES BLOB,
+	REGION VARCHAR(100) NOT NULL,
+	constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
+) ENGINE=InnoDB;
+
+ALTER TABLE INT_CHANNEL_MESSAGE
+ADD INDEX MSG_INDEX_DATE_IDX USING BTREE (CREATED_DATE ASC) ;
--- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-oracle.sql
+++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-oracle.sql
@@ -0,0 +1,11 @@
+CREATE TABLE INT_CHANNEL_MESSAGE (
+	MESSAGE_ID CHAR(36) NOT NULL,
+	GROUP_KEY CHAR(36) NOT NULL,
+	CREATED_DATE NUMBER(19,0) NOT NULL,
+	MESSAGE_BYTES BLOB,
+	REGION VARCHAR2(100) NOT NULL,
+	constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
+);
+
+CREATE INDEX INT_CHANNEL_MSG_DATE_IDX
+	ON INT_CHANNEL_MESSAGE (CREATED_DATE);
--- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-postgresql.sql
+++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-postgresql.sql
@@ -0,0 +1,12 @@
+CREATE TABLE INT_CHANNEL_MESSAGE (
+	MESSAGE_ID character(36) NOT NULL,
+	GROUP_KEY character(36) NOT NULL,
+	CREATED_DATE BIGINT NOT NULL,
+	MESSAGE_BYTES bytea,
+	REGION character varying(100) NOT NULL,
+	constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
+);
+
+CREATE INDEX MSG_INDEX_DATE_IDX
+	ON INT_CHANNEL_MESSAGE
+	USING btree (created_date);
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageHandlerIntegrationTests.java
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageHandlerIntegrationTests.java
@@ -58,7 +58,7 @@ public class JdbcMessageHandlerIntegrationTests {
 	public void testSimpleStaticInsert() {
 		JdbcMessageHandler handler = new JdbcMessageHandler(jdbcTemplate, "insert into foos (id, status, name) values (1, 0, 'foo')");
 		Message<String> message = new GenericMessage<String>("foo");
-		handler.handleMessage(message);		
+		handler.handleMessage(message);
 		Map<String, Object> map = jdbcTemplate.queryForMap("SELECT * FROM FOOS WHERE ID=?", 1);
 		assertEquals("Wrong id", "1", map.get("ID"));
 		assertEquals("Wrong status", 0, map.get("STATUS"));
@@ -73,7 +73,7 @@ public class JdbcMessageHandlerIntegrationTests {
 		Map<String, Object> map = jdbcTemplate.queryForMap("SELECT * FROM FOOS WHERE ID=?", 1);
 		assertEquals("Wrong name", "foo", map.get("NAME"));
 	}
-	
+
 	@Test
 	public void testIdHeaderDynamicInsert() {
 		JdbcMessageHandler handler = new JdbcMessageHandler(jdbcTemplate, "insert into foos (id, status, name) values (:headers[id], 0, :payload)");
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcTypesEnumTests.java
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcTypesEnumTests.java
@@ -12,46 +12,46 @@ public class JdbcTypesEnumTests {

 	@Test
 	public void testGetCode() {
-		
+
 		JdbcTypesEnum jdbcTypesEnum = JdbcTypesEnum.convertToJdbcTypesEnum("VARCHAR");
 		assertNotNull("Expected not null jdbcTypesEnum.", jdbcTypesEnum);
 		assertEquals(Integer.valueOf(Types.VARCHAR), Integer.valueOf(jdbcTypesEnum.getCode()));
-		
+
 	}

 	@Test
 	public void testConvertToJdbcTypesEnumWithInvalidParameter() {
-		
+
 		JdbcTypesEnum jdbcTypesEnum = JdbcTypesEnum.convertToJdbcTypesEnum("KENNY4JDBC");
 		assertNull("Expected null return value.", jdbcTypesEnum);
-		
+
 	}
-	
+
 	@Test
 	public void testConvertToJdbcTypesEnumWithNullParameter() {
-		
+
 		try {
 		    JdbcTypesEnum.convertToJdbcTypesEnum(null);
 		} catch (IllegalArgumentException e) {
 			assertEquals("Parameter sqlTypeAsString, must not be null nor empty", e.getMessage());
 			return;
 		}
-		
+
 		fail("Expected Exception");
-		
+
 	}

 	@Test
 	public void testConvertToJdbcTypesEnumWithEmptyParameter() {
-		
+
 		try {
 		    JdbcTypesEnum.convertToJdbcTypesEnum("   ");
 		} catch (IllegalArgumentException e) {
 			assertEquals("Parameter sqlTypeAsString, must not be null nor empty", e.getMessage());
 			return;
 		}
-		
+
 		fail("Expected Exception");
-		
+
 	}
 }
--- 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
@@ -0,0 +1,93 @@
+/*
+ * Copyright 2002-2012 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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
+import javax.sql.DataSource;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.integration.Message;
+import org.springframework.integration.jdbc.store.JdbcChannelMessageStore;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.TransactionStatus;
+import org.springframework.transaction.annotation.Isolation;
+import org.springframework.transaction.support.TransactionCallbackWithoutResult;
+import org.springframework.transaction.support.TransactionTemplate;
+
+/**
+ * @author Gunnar Hillert
+ */
+public class AbstractJdbcChannelMessageStoreTests {
+
+	protected static final String TEST_MESSAGE_GROUP = "AbstractJdbcChannelMessageStoreTests";
+
+	@Autowired
+	protected DataSource dataSource;
+
+	protected JdbcChannelMessageStore messageStore;
+
+	@Autowired
+	protected PlatformTransactionManager transactionManager;
+
+	@Autowired
+	protected ChannelMessageStoreQueryProvider queryProvider;
+
+	public void init() throws Exception {
+		messageStore = new JdbcChannelMessageStore(dataSource);
+		messageStore.setRegion("AbstractJdbcChannelMessageStoreTests");
+		messageStore.setChannelMessageStoreQueryProvider(queryProvider);
+		messageStore.afterPropertiesSet();
+		messageStore.removeMessageGroup("AbstractJdbcChannelMessageStoreTests");
+	}
+
+	public void testGetNonExistentMessageFromGroup() throws Exception {
+		Message<?> result = messageStore.pollMessageFromGroup(TEST_MESSAGE_GROUP);
+		assertNull(result);
+	}
+
+	public void testAddAndGet() throws Exception {
+		final Message<String> message = MessageBuilder.withPayload("Cartman and Kenny")
+				.setHeader("homeTown", "Southpark")
+				.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());
+
+		assertNotNull(messageFromDb.getHeaders().get(JdbcChannelMessageStore.SAVED_KEY));
+		assertNotNull(messageFromDb.getHeaders().get(JdbcChannelMessageStore.CREATED_DATE_KEY));
+	}
+
+}
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractTxTimeoutMessageStoreTests.java
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractTxTimeoutMessageStoreTests.java
@@ -0,0 +1,92 @@
+/*
+ * Copyright 2002-2012 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 javax.sql.DataSource;
+
+import junit.framework.Assert;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.integration.MessageChannel;
+import org.springframework.integration.jdbc.store.JdbcChannelMessageStore;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.TransactionStatus;
+import org.springframework.transaction.annotation.Isolation;
+import org.springframework.transaction.support.TransactionCallbackWithoutResult;
+import org.springframework.transaction.support.TransactionTemplate;
+
+/**
+ *
+ * @author Gunnar Hillert
+ *
+ */
+abstract class AbstractTxTimeoutMessageStoreTests {
+
+	private static final Log log = LogFactory.getLog(AbstractTxTimeoutMessageStoreTests.class);
+
+	@Autowired
+	protected DataSource dataSource;
+
+	@Autowired
+	protected MessageChannel inputChannel;
+
+	@Autowired
+	protected PlatformTransactionManager transactionManager;
+
+	@Autowired
+	protected TestService testService;
+
+	@Autowired
+	protected JdbcChannelMessageStore jdbcChannelMessageStore;
+
+	public void test() throws InterruptedException {
+
+		int maxMessages = 10;
+		int maxWaitTime = 30000;
+
+		final TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
+
+		transactionTemplate.setIsolationLevel(Isolation.READ_COMMITTED.value());
+		transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_NEVER);
+
+		for (int i = 1; i <= maxMessages; ++i) {
+			final String message = "TEST MESSAGE " + i;
+			log.info("Sending message: " + message);
+
+			transactionTemplate.execute(new TransactionCallbackWithoutResult() {
+				@Override
+				protected void doInTransactionWithoutResult(TransactionStatus status) {
+					inputChannel.send(MessageBuilder.withPayload(message).build());
+				}
+			});
+
+			log.info(String.format("Done sending message %s of %s: %s", i, maxMessages, message));
+		}
+
+		log.info("Done sending " + maxMessages + " messages.");
+
+		Assert.assertTrue(String.format("Contdown latch did not count down from " +
+				"%s to 0 in %sms.", maxMessages, maxWaitTime), testService.await(maxWaitTime));
+
+		Thread.sleep(2000);
+
+		Assert.assertEquals(Integer.valueOf(0), Integer.valueOf(jdbcChannelMessageStore.getSizeOfIdCache()));
+		Assert.assertEquals(Integer.valueOf(maxMessages), Integer.valueOf(testService.getSeenMessages().size()));
+		Assert.assertEquals(Integer.valueOf(0), Integer.valueOf(testService.getDuplicateMessagesCount()));
+	}
+
+}
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/DataSource-common-context.xml
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/DataSource-common-context.xml
@@ -0,0 +1,16 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:jdbc="http://www.springframework.org/schema/jdbc"
+	xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
+			http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
+
+	<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
+		<property name="dataSource" ref="dataSource" />
+		<!-- <property name="defaultTimeout" value="2000" /> -->
+		<!-- Postgres 9 does not support send timeouts. You may see an exception like:
+		"java.sql.SQLFeatureNotSupportedException: Method org.postgresql.jdbc4.Jdbc4PreparedStatement.setQueryTimeout(int) is not yet implemented."
+		-->
+	</bean>
+
+</beans>
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/DataSource-derby-context.xml
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/DataSource-derby-context.xml
@@ -0,0 +1,34 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:jdbc="http://www.springframework.org/schema/jdbc"
+	xsi:schemaLocation="
+		http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
+
+	<import resource="classpath:org/springframework/integration/jdbc/store/channel/DataSource-common-context.xml" />
+
+	<bean id="dataSource" class="org.apache.commons.dbcp.BasicDataSource"
+		destroy-method="close">
+		<property name="driverClassName" value="org.apache.derby.jdbc.EmbeddedDriver" />
+		<property name="initialSize" value="10" />
+		<property name="url" value="jdbc:derby:memory:integration;create=true" />
+		<property name="username" value="int" />
+		<property name="password" value="int" />
+	</bean>
+
+	<!-- <bean id="dataSource" class="org.apache.commons.dbcp.BasicDataSource"
+		destroy-method="close"> <property name="driverClassName" value="org.apache.derby.jdbc.ClientDriver"/>
+		<property name="initialSize" value="10"/> <property name="url" value="jdbc:derby://localhost:1527/integration;create=true"
+		/> <property name="username" value="int" /> <property name="password" value="int"
+		/> </bean> -->
+
+	<jdbc:initialize-database data-source="dataSource">
+		<jdbc:script
+			location="classpath:org/springframework/integration/jdbc/store/channel/schema-derby.sql" />
+	</jdbc:initialize-database>
+
+	<bean id="queryProvider"
+		class="org.springframework.integration.jdbc.store.channel.DerbyChannelMessageStoreQueryProvider" />
+
+</beans>
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/DataSource-hsql-context.xml
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/DataSource-hsql-context.xml
@@ -0,0 +1,25 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:jdbc="http://www.springframework.org/schema/jdbc"
+	xsi:schemaLocation="
+		http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
+
+	<import resource="classpath:org/springframework/integration/jdbc/store/channel/DataSource-common-context.xml" />
+
+	<bean id="dataSource" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">
+		<property name="driverClassName" value="org.hsqldb.jdbcDriver"/>
+		<property name="initialSize" value="10"/>
+		<property name="url" value="jdbc:hsqldb:mem:integration;hsqldb.tx=mvcc" />
+		<property name="username" value="sa" />
+		<property name="password" value="" />
+	</bean>
+
+	<jdbc:initialize-database data-source="dataSource" ignore-failures="DROPS">
+		<jdbc:script execution="INIT" location="classpath:org/springframework/integration/jdbc/store/channel/schema-drop-hsql.sql"/>
+		<jdbc:script execution="INIT" location="classpath:org/springframework/integration/jdbc/store/channel/schema-hsql.sql"/>
+	</jdbc:initialize-database>
+
+	<bean id="queryProvider" class="org.springframework.integration.jdbc.store.channel.HsqlChannelMessageStoreQueryProvider"/>
+</beans>
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/DataSource-mysql-context.xml
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/DataSource-mysql-context.xml
@@ -0,0 +1,22 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:jdbc="http://www.springframework.org/schema/jdbc"
+	xsi:schemaLocation="
+		http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
+
+	<import resource="classpath:org/springframework/integration/jdbc/store/channel/DataSource-common-context.xml" />
+
+	<bean id="dataSource" destroy-method="close"
+		class="org.apache.commons.dbcp.BasicDataSource">
+		<property name="driverClassName" value="com.mysql.jdbc.Driver" />
+		<property name="url" value="jdbc:mysql://localhost/test" />
+		<property name="username" value="root" />
+		<property name="password" value="" />
+		<property name="initialSize" value="10" />
+	</bean>
+
+	<bean id="queryProvider" class="org.springframework.integration.jdbc.store.channel.MySqlChannelMessageStoreQueryProvider"/>
+
+</beans>
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/DataSource-oracle-context.xml
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/DataSource-oracle-context.xml
@@ -0,0 +1,26 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:jdbc="http://www.springframework.org/schema/jdbc"
+	xsi:schemaLocation="
+		http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
+
+	<import resource="classpath:org/springframework/integration/jdbc/store/channel/DataSource-common-context.xml" />
+
+	<bean id="dataSource" class="oracle.jdbc.pool.OracleDataSource" destroy-method="close">
+		<property name="connectionCachingEnabled" value="true" />
+		<property name="URL" value="jdbc:oracle:thin:@//10.0.1.6:1521/XE" />
+		<property name="password" value="TT2" />
+		<property name="user"     value="tt" />
+		<property name="connectionCacheProperties">
+			<props merge="default">
+				 <prop key="MinLimit">3</prop>
+				 <prop key="MaxLimit">20</prop>
+			</props>
+		 </property>
+	</bean>
+
+	<bean id="queryProvider" class="org.springframework.integration.jdbc.store.channel.OracleChannelMessageStoreQueryProvider"/>
+
+</beans>
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/DataSource-postgres-context.xml
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/DataSource-postgres-context.xml
@@ -0,0 +1,23 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:jdbc="http://www.springframework.org/schema/jdbc"
+	xsi:schemaLocation="
+		http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
+
+	<import resource="classpath:org/springframework/integration/jdbc/store/channel/DataSource-common-context.xml" />
+
+	<bean id="dataSource" class="org.apache.commons.dbcp.BasicDataSource"
+		destroy-method="close">
+		<property name="driverClassName" value="org.postgresql.Driver" />
+		<property name="initialSize" value="10"/>
+		<property name="url"
+			value="jdbc:postgresql:integration" />
+		<property name="username" value="postgres" />
+		<property name="password" value="postgres" />
+	</bean>
+
+	<bean id="queryProvider" class="org.springframework.integration.jdbc.store.channel.PostgresChannelMessageStoreQueryProvider"/>
+
+</beans>
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/DerbyTxTimeoutMessageStoreTests-context.xml
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/DerbyTxTimeoutMessageStoreTests-context.xml
@@ -0,0 +1,20 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xmlns:jdbc="http://www.springframework.org/schema/jdbc"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc"
+	xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
+		http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/task/spring-jdbc.xsd
+		http://www.springframework.org/schema/integration/jdbc http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
+
+	<import resource="classpath:org/springframework/integration/jdbc/store/channel/DataSource-derby-context.xml"/>
+	<import resource="classpath:org/springframework/integration/jdbc/store/channel/TxTimeoutMessageStoreTests-context.xml"/>
+
+</beans>
+
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/DerbyTxTimeoutMessageStoreTests.java
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/DerbyTxTimeoutMessageStoreTests.java
@@ -0,0 +1,39 @@
+/*
+ * Copyright 2002-2012 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 org.junit.Ignore;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+
+
+/**
+ *
+ * @author Gunnar Hillert
+ *
+ */
+@Ignore
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class DerbyTxTimeoutMessageStoreTests extends AbstractTxTimeoutMessageStoreTests {
+
+	@Test
+	@Override
+	public void test() throws InterruptedException {
+		super.test();
+	}
+
+}
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/HsqlJdbcChannelMessageStoreTests-context.xml
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/HsqlJdbcChannelMessageStoreTests-context.xml
@@ -0,0 +1,11 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:jdbc="http://www.springframework.org/schema/jdbc"
+	xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
+			http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
+
+	<import resource="classpath:org/springframework/integration/jdbc/store/channel/DataSource-hsql-context.xml"/>
+	<import resource="classpath:org/springframework/integration/jdbc/store/channel/JdbcChannelMessageStoreTests-context.xml"/>
+
+</beans>
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/HsqlJdbcChannelMessageStoreTests.java
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/HsqlJdbcChannelMessageStoreTests.java
@@ -0,0 +1,50 @@
+/*
+ * Copyright 2002-2012 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 org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Gunnar Hillert
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class HsqlJdbcChannelMessageStoreTests extends AbstractJdbcChannelMessageStoreTests {
+
+	@Before
+	@Override
+	public void init() throws Exception {
+		super.init();
+	}
+
+	@Test
+	@Override
+	public void testGetNonExistentMessageFromGroup() throws Exception {
+		super.testGetNonExistentMessageFromGroup();
+	}
+
+	@Test
+	@Override
+	public void testAddAndGet() throws Exception {
+		super.testAddAndGet();
+	}
+
+}
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/HsqlTxTimeoutMessageStoreTests-context.xml
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/HsqlTxTimeoutMessageStoreTests-context.xml
@@ -0,0 +1,20 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xmlns:jdbc="http://www.springframework.org/schema/jdbc"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc"
+	xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
+		http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/task/spring-jdbc.xsd
+		http://www.springframework.org/schema/integration/jdbc http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
+
+	<import resource="classpath:org/springframework/integration/jdbc/store/channel/DataSource-hsql-context.xml"/>
+	<import resource="classpath:org/springframework/integration/jdbc/store/channel/TxTimeoutMessageStoreTests-context.xml"/>
+
+</beans>
+
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/HsqlTxTimeoutMessageStoreTests.java
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/HsqlTxTimeoutMessageStoreTests.java
@@ -0,0 +1,37 @@
+/*
+ * Copyright 2002-2012 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 org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ *
+ * @author Gunnar Hillert
+ *
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class HsqlTxTimeoutMessageStoreTests extends AbstractTxTimeoutMessageStoreTests {
+
+	@Test
+	@Override
+	public void test() throws InterruptedException {
+
+		super.test();
+
+	}
+
+}
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/JdbcChannelMessageStoreTests-context.xml
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/JdbcChannelMessageStoreTests-context.xml
@@ -0,0 +1,8 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:jdbc="http://www.springframework.org/schema/jdbc"
+	xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
+			http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
+
+</beans>
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/MySqlJdbcChannelMessageStoreTests-context.xml
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/MySqlJdbcChannelMessageStoreTests-context.xml
@@ -0,0 +1,11 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:jdbc="http://www.springframework.org/schema/jdbc"
+	xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
+			http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
+
+	<import resource="classpath:org/springframework/integration/jdbc/store/channel/DataSource-mysql-context.xml"/>
+	<import resource="classpath:org/springframework/integration/jdbc/store/channel/JdbcChannelMessageStoreTests-context.xml"/>
+
+</beans>
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/MySqlJdbcChannelMessageStoreTests.java
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/MySqlJdbcChannelMessageStoreTests.java
@@ -0,0 +1,53 @@
+/*
+ * Copyright 2002-2012 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 org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+
+/**
+ * @author Gunnar Hillert
+ */
+@Ignore
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class MySqlJdbcChannelMessageStoreTests extends AbstractJdbcChannelMessageStoreTests {
+
+	@Before
+	@Override
+	public void init() throws Exception {
+		super.init();
+	}
+
+	@Test
+	@Override
+	public void testGetNonExistentMessageFromGroup() throws Exception {
+		super.testGetNonExistentMessageFromGroup();
+	}
+
+	@Test
+	@Override
+	public void testAddAndGet() throws Exception {
+		super.testAddAndGet();
+	}
+
+}
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/MySqlTxTimeoutMessageStoreTests-context.xml
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/MySqlTxTimeoutMessageStoreTests-context.xml
@@ -0,0 +1,20 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xmlns:jdbc="http://www.springframework.org/schema/jdbc"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc"
+	xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
+		http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/task/spring-jdbc.xsd
+		http://www.springframework.org/schema/integration/jdbc http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
+
+	<import resource="classpath:org/springframework/integration/jdbc/store/channel/DataSource-mysql-context.xml"/>
+	<import resource="classpath:org/springframework/integration/jdbc/store/channel/TxTimeoutMessageStoreTests-context.xml"/>
+
+</beans>
+
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/MySqlTxTimeoutMessageStoreTests.java
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/MySqlTxTimeoutMessageStoreTests.java
@@ -0,0 +1,38 @@
+/*
+ * Copyright 2002-2012 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 org.junit.Ignore;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+
+/**
+ *
+ * @author Gunnar Hillert
+ *
+ */
+@Ignore
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class MySqlTxTimeoutMessageStoreTests extends AbstractTxTimeoutMessageStoreTests {
+
+	@Test
+	@Override
+	public void test() throws InterruptedException {
+		super.test();
+	}
+
+}
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/OracleTxTimeoutMessageStoreTests-context.xml
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/OracleTxTimeoutMessageStoreTests-context.xml
@@ -0,0 +1,20 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xmlns:jdbc="http://www.springframework.org/schema/jdbc"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc"
+	xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
+		http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/task/spring-jdbc.xsd
+		http://www.springframework.org/schema/integration/jdbc http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
+
+	<import resource="classpath:org/springframework/integration/jdbc/store/channel/DataSource-oracle-context.xml"/>
+	<import resource="classpath:org/springframework/integration/jdbc/store/channel/TxTimeoutMessageStoreTests-context.xml"/>
+
+</beans>
+
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/OracleTxTimeoutMessageStoreTests.java
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/OracleTxTimeoutMessageStoreTests.java
@@ -0,0 +1,38 @@
+/*
+ * Copyright 2002-2012 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 org.junit.Ignore;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+
+/**
+ *
+ * @author Gunnar Hillert
+ *
+ */
+@Ignore
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class OracleTxTimeoutMessageStoreTests extends AbstractTxTimeoutMessageStoreTests {
+
+	@Test
+	@Override
+	public void test() throws InterruptedException {
+		super.test();
+	}
+
+}
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/PostgresTxTimeoutMessageStoreTests-context.xml
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/PostgresTxTimeoutMessageStoreTests-context.xml
@@ -0,0 +1,20 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xmlns:jdbc="http://www.springframework.org/schema/jdbc"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc"
+	xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
+		http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/task/spring-jdbc.xsd
+		http://www.springframework.org/schema/integration/jdbc http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
+
+	<import resource="classpath:org/springframework/integration/jdbc/store/channel/DataSource-postgres-context.xml"/>
+	<import resource="classpath:org/springframework/integration/jdbc/store/channel/TxTimeoutMessageStoreTests-context.xml"/>
+
+</beans>
+
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/PostgresTxTimeoutMessageStoreTests.java
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/PostgresTxTimeoutMessageStoreTests.java
@@ -0,0 +1,38 @@
+/*
+ * Copyright 2002-2012 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 org.junit.Ignore;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+
+/**
+ *
+ * @author Gunnar Hillert
+ *
+ */
+@Ignore
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class PostgresTxTimeoutMessageStoreTests extends AbstractTxTimeoutMessageStoreTests {
+
+	@Test
+	@Override
+	public void test() throws InterruptedException {
+		super.test();
+	}
+
+}
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/TestService.java
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/TestService.java
@@ -0,0 +1,89 @@
+/*
+ * Copyright 2002-2012 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.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+/**
+ *
+ * @author Gunnar Hillert
+ *
+ */
+public class TestService {
+
+	private static final Log log = LogFactory.getLog(TestService.class);
+
+	private Map<String, String> seen = new ConcurrentHashMap<String, String>();
+
+	private AtomicInteger duplicateMessagesCount = new AtomicInteger(0);
+
+	private final CountDownLatch latch;
+
+	private int maxNumberOfMessages = 10;
+	private int threadSleep = 10;
+
+	public TestService(int maxNumberOfMessages, int threadSleep) {
+		super();
+		this.maxNumberOfMessages = maxNumberOfMessages;
+		latch = new CountDownLatch(maxNumberOfMessages);
+		this.threadSleep = threadSleep;
+	}
+
+	public TestService() {
+		super();
+		this.maxNumberOfMessages = 10;
+		latch = new CountDownLatch(maxNumberOfMessages);
+	}
+
+	public void process(final String message) {
+		log.info("Received: " + message);
+
+		if (seen.containsKey(message)) {
+			log.error("Already seen: " + message);
+			duplicateMessagesCount.addAndGet(1);
+		} else {
+			seen.put(message, message);
+			log.info("Pre: " + message);
+
+			if (this.threadSleep >0) {
+				try {
+					Thread.sleep(2000);
+				} catch (InterruptedException e) {
+					Thread.currentThread().interrupt();
+				}
+			}
+			log.info("Post: " + message + "; latch: " + latch.getCount());
+			latch.countDown();
+		}
+	}
+
+	public boolean await(int milliseconds) throws InterruptedException {
+		return latch.await(milliseconds, TimeUnit.MILLISECONDS);
+	}
+
+	public Map<String, String> getSeenMessages() {
+		return seen;
+	}
+
+	public int getDuplicateMessagesCount() {
+		return duplicateMessagesCount.get();
+	}
+
+}
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/TxTimeoutMessageStoreTests-context.xml
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/TxTimeoutMessageStoreTests-context.xml
@@ -0,0 +1,58 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xmlns:jdbc="http://www.springframework.org/schema/jdbc"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc"
+	xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
+		http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/task/spring-jdbc.xsd
+		http://www.springframework.org/schema/integration/jdbc http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
+
+	<int:transaction-synchronization-factory id="syncFactory">
+		<int:after-commit expression="@store.removeFromIdCache(headers.id.toString())" />
+		<int:after-rollback expression="@store.removeFromIdCache(headers.id.toString())"/>
+	</int:transaction-synchronization-factory>
+
+	<task:executor id="pool" pool-size="10" queue-capacity="10" rejection-policy="CALLER_RUNS" />
+
+	<bean id="store" class="org.springframework.integration.jdbc.store.JdbcChannelMessageStore">
+		<property name="dataSource" ref="dataSource"/>
+		<property name="channelMessageStoreQueryProvider" ref="queryProvider"/>
+		<property name="region" value="TX_TIMEOUT"/>
+		<property name="usingIdCache" value="true"/>
+	</bean>
+
+	<int:channel id="inputChannel">
+		<int:queue message-store="store"/>
+	</int:channel>
+
+	<int:bridge input-channel="inputChannel" output-channel="outputChannel">
+		<int:poller fixed-delay="500" receive-timeout="500"
+			max-messages-per-poll="1" task-executor="pool">
+			<int:transactional propagation="REQUIRED" synchronization-factory="syncFactory"
+				isolation="READ_COMMITTED" transaction-manager="transactionManager" />
+		</int:poller>
+	</int:bridge>
+
+	<int:channel id="outputChannel" />
+
+	<bean id="testService" class="org.springframework.integration.jdbc.store.channel.TestService"/>
+
+	<int:service-activator input-channel="outputChannel" ref="testService" method="process"/>
+
+	<int:channel id="errorChannel">
+		<int:queue/>
+		<int:interceptors>
+			<int:wire-tap channel="loggit"/>
+		</int:interceptors>
+	</int:channel>
+
+	<int:logging-channel-adapter id="loggit" log-full-message="true" level="ERROR"/>
+
+</beans>
+
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/storedproc/h2/H2StoredProcedures.java
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/storedproc/h2/H2StoredProcedures.java
@@ -3,12 +3,12 @@ package org.springframework.integration.jdbc.storedproc.h2;
 import java.math.BigInteger;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.sql.Types;

 import org.h2.tools.SimpleResultSet;
-import org.hsqldb.Types;

 /**
- * 
+ *
 * @author Gunnar Hillert
 *
 */
--- a/spring-integration-jdbc/src/test/resources/log4j.properties
+++ b/spring-integration-jdbc/src/test/resources/log4j.properties
@@ -6,3 +6,6 @@ log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{2}:%L - %m

 log4j.category.org.springframework.integration=INFO
 log4j.category.org.apache.derby=INFO
+
+log4j.category.org.springframework.jdbc=WARN
+log4j.category.org.springframework.integration.jdbc.JdbcChannelMessageStore=INFO
--- a/src/reference/docbook/jdbc.xml
+++ b/src/reference/docbook/jdbc.xml
@@ -297,38 +297,149 @@

  <section id="jdbc-message-store">
    <title>JDBC Message Store</title>
-
-    <para>The JDBC module provides an implementation of the Spring Integration
-    <classname>MessageStore</classname> (important in the Claim Check pattern)
-    and <classname>MessageGroupStore</classname> (important in stateful
-    patterns like Aggregator) backed by a database. Both interfaces are
-    implemented by the JdbcMessageStore, and there is also support for
-    configuring store instances in XML. For example:</para>
-
-    <programlisting language="xml"><![CDATA[<int-jdbc:message-store id="messageStore" data-source="dataSource"/>
+	<para>
+		Spring Integration provides 2 JDBC specifc Message Store implementations.
+		The first one, is the <classname>JdbcMessageStore</classname> which is
+		suitable to be used in conjunction with <emphasis>Aggregators</emphasis>
+		and the <emphasis>Claimcheck</emphasis> pattern. While it can be used for
+		backing <emphasis>Message Channels</emphasis> as well, you may want to
+		consider using the <classname>JdbcChannelMessageStore</classname> implementation
+		instead, as it provides a more targeted and scalable implementation.
+	</para>
+	<section id="jdbc-message-store-generic">
+		<title>The Generic JDBC Message Store</title>
+		<para>
+			The JDBC module provides an implementation of the Spring Integration
+			<classname>MessageStore</classname> (important in the Claim Check pattern)
+ 			and <classname>MessageGroupStore</classname> (important in stateful
+			patterns like Aggregator) backed by a database. Both interfaces are
+			implemented by the <classname>JdbcMessageStore</classname>, and there
+			is also support for configuring store instances in XML.
+			For example:</para>
+			<programlisting language="xml"><![CDATA[<int-jdbc:message-store id="messageStore" data-source="dataSource"/>
 ]]></programlisting>

-    <para>A <classname>JdbcTemplate</classname> can be specified instead of a
-    <classname>DataSource</classname>.</para>
-
-    <para>Other optional attributes are show in the next example:</para>
-
-    <para><programlisting language="xml"><![CDATA[<int-jdbc:message-store id="messageStore" data-source="dataSource"
+		<para>
+			A <classname>JdbcTemplate</classname> can be specified instead of a
+			<classname>DataSource</classname>.
+		</para>
+		<para>
+			Other optional attributes are show in the next example:
+		</para>
+		<programlisting language="xml"><![CDATA[<int-jdbc:message-store id="messageStore" data-source="dataSource"
    lob-handler="lobHandler" table-prefix="MY_INT_"/>]]></programlisting>
+		<para>
+			Here we have specified a <classname>LobHandler</classname> for dealing
+			with messages as large objects (e.g. often necessary if using Oracle)
+			and a prefix for the table names in the queries generated by the store.
+			The table name prefix defaults to "INT_".
+		</para>
+	</section>
+	<section id="jdbc-message-store-channels">
+		<title>Backing Message Channels</title>
+		<para>
+			If you intent backing <emphasis>Message Channels</emphasis> using JDBC,
+			it is recommended to use the provided <classname>JdbcChannelMessageStore</classname>
+			implementation instead. It can only be used in conjuntion
+			with <emphasis>Message Channels</emphasis>.
+		</para>
+		<note>
+			The provided <classname>JdbcChannelMessageStore</classname>
+			implementation is available since <emphasis>Spring Integration 2.2.</emphasis>.
+		</note>
+		<para><emphasis>Supported Database</emphasis></para>
+		<para>
+			The <classname>JdbcChannelMessageStore</classname> uses database specific
+			SQL queries to retrieve messages from the database. Therefore, users must
+			set the <interfacename>ChannelMessageStoreQueryProvider</interfacename> property on the
+			<classname>JdbcChannelMessageStore</classname>. This
+			<interfacename>channelMessageStoreQueryProvider</interfacename>
+			provides the SQL queries and Spring Integration provides support for
+			the following relational databases:
+		</para>
+		<itemizedlist>
+			<listitem>PostgreSQL</listitem>
+			<listitem>HSQLDB</listitem>
+			<listitem>MySQL</listitem>
+			<listitem>Oracle</listitem>
+			<listitem>Derby</listitem>
+		</itemizedlist>
+		<para>
+			If your database is not listed, you can easily extend
+			the <classname>AbstractChannelMessageStoreQueryProvider</classname>
+			class and provide your own custom queries.
+		</para>
+		<important>
+			<para>
+				Generally it is not recommened to use a relational database for the
+				purpose of queuing. Instead, if possible, consider using either JMS or
+				AMQP, for which message store implementation are provided as well. For
+				further reference please see the following resources:
+			</para>
+			<itemizedlist>
+				<listitem><ulink url="https://www.engineyard.com/blog/2011/5-subtle-ways-youre-using-mysql-as-a-queue-and-why-itll-bite-you/">5 subtle ways you’re using MySQL as a queue, and why it’ll bite you</ulink></listitem>
+				<listitem><ulink url="http://mikehadlow.blogspot.com/2012/04/database-as-queue-anti-pattern.html">The Database As Queue Anti-Pattern</ulink></listitem>
+			</itemizedlist>
+		</important>
+		<para><emphasis>Concurrent Polling</emphasis></para>
+		<para>
+			When polling a <emphasis>Message Channel</emphasis>, you have the option
+			to configure the associated <classname>Poller</classname> with a
+			<interfacename>TaskExecutor</interfacename> reference.
+		</para>
+		<para>
+			Keep in mind, though, that if you use a JDBC backed <emphasis>Message Channel</emphasis> and
+			you are planning on polling the channel and consequently the message
+			store transactionally with multiple threads, you should ensure that
+			you use a relational database that supports
+			<ulink url="http://en.wikipedia.org/wiki/Multiversion_concurrency_control">Multiversion Concurrency Control</ulink> (MVCC).
+			Otherwise, locking may be an issue and the performance, when using multiple
+			threads, may not materialize as expected. For example Apache Derby is
+			problematic in that regard.
+		</para>
+		<programlisting language="xml"><![CDATA[…
+<bean id="queryProvider"
+    class="o.s.i.jdbc.store.channel.PostgresChannelMessageStoreQueryProvider"/>

-    Here we have specified a <classname>LobHandler</classname> for dealing with
-    messages as large objects (e.g. often necessary if using Oracle) and a
-    prefix for the table names in the queries generated by the store. The
-    table name prefix defaults to "INT_".</para>
+<int:transaction-synchronization-factory id="syncFactory">
+    <int:after-commit expression="@store.removeFromIdCache(headers.id.toString())" />
+    <int:after-rollback expression="@store.removeFromIdCache(headers.id.toString())"/>
+</int:transaction-synchronization-factory>

+<task:executor id="pool" pool-size="10"
+    queue-capacity="10" rejection-policy="CALLER_RUNS" />
+
+<bean id="store" class="o.s.i.jdbc.store.JdbcChannelMessageStore">
+    <property name="dataSource" ref="dataSource"/>
+    <property name="channelMessageStoreQueryProvider" ref="queryProvider"/>
+    <property name="region" value="TX_TIMEOUT"/>
+    <property name="usingIdCache" value="true"/>
+</bean>
+
+<int:channel id="inputChannel">
+    <int:queue message-store="store"/>
+</int:channel>
+
+<int:bridge input-channel="inputChannel" output-channel="outputChannel">
+    <int:poller fixed-delay="500" receive-timeout="500"
+        max-messages-per-poll="1" task-executor="pool">
+        <int:transactional propagation="REQUIRED" synchronization-factory="syncFactory"
+        isolation="READ_COMMITTED" transaction-manager="transactionManager" />
+    </int:poller>
+</int:bridge>
+
+<int:channel id="outputChannel" />
+…]]></programlisting>
+	</section>
    <section>
      <title>Initializing the Database</title>

      <para>Spring Integration ships with some sample scripts that can be used
      to initialize a database. In the spring-integration-jdbc JAR file you
      will find scripts in the
-      <classname>org.springframework.integration.jdbc</classname> package:
-      there is a create and a drop script example for a range of common
+      <classname>org.springframework.integration.jdbc</classname> and in the
+		<classname>org.springframework.integration.jdbc.store.channel</classname>
+		package: there is a create and a drop script example for a range of common
      database platforms. A common way to use these scripts is to reference
      them in a <ulink
      url="http://static.springsource.org/spring/docs/3.0.x/spring-framework-reference/html/jdbc.html#d0e24182">Spring
@@ -353,7 +464,7 @@
      message data for a persistent channel is keyed in the store on the
      channel name, so if the channel names are not globally unique then there
      is the danger of channels picking up data that was not intended for
-      them. To avoid this, the message store region can be used to keep data
+      them. To avoid this, the message store <emphasis>region</emphasis> can be used to keep data
      separate for different physical channels that happen to have the same
      logical name.</para>
    </section>
--- a/src/reference/docbook/whats-new.xml
+++ b/src/reference/docbook/whats-new.xml
@@ -67,7 +67,7 @@
 			</para>
 		</section>
 		<section id="2.2-jdbc-11">
-			<title>JDBC Adapter - Stored Procedures Components</title>
+			<title>JDBC Support - Stored Procedures Components</title>
 			<para><emphasis>SpEL Support</emphasis></para>
 			<para>
 				When using the Stored Procedure components of the Spring Integration
@@ -92,13 +92,22 @@
 			</itemizedlist>
 		</section>
 		<section id="2.2-jdbc-gateway-update-optional">
-			<title>JDBC Adapter - Outbound Gateway</title>
+			<title>JDBC Support - Outbound Gateway</title>
 			<para>
 				When using the JDBC Outbound Gateway, the update query is no longer
 				mandatory. You can now provide solely a select query using the request
 				message as a source of parameters.
 			</para>
 		</section>
+		<section id="2.2-jdbc-message-store-channels">
+			<title>JDBC Support - Channel-specific Message Store Implementation</title>
+			<para>
+				A new <emphasis>Message Channel</emphasis>-specific Message Store
+				Implementation has been added, providing a more scalable solution
+				using database-specific SQL queries. For more information please
+				see: <xref linkend="jdbc-message-store-channels"/>.
+			</para>
+		</section>
 		<section id="2.2-file-adapter">
 			<title>File Adapter - Improved File Overwrite/Append Handling</title>
 			<para>