+ * 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:
+ *
+ *
+ * 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.
+ *
+ *
+ * @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 ) 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;
+ }
+
+ /**
+ * Consider using this property when polling the database transactionally
+ * using multiple parallel threads, meaning when the configured poller is configured
+ * using a task executor.
+ *
+ * The issue is that the {@link #pollMessageFromGroup(Object)} looks for the
+ * oldest entry for a giving channel (groupKey) and region ({@link #setRegion(String)}).
+ * If you do that with multiple threads and you are using transactions, other
+ * threads may be waiting for that same locked row.
+ *
+ * If using the provided {@link OracleChannelMessageStoreQueryProvider}, don't set {@link #usingIdCache}
+ * to true, as the Oracle query will ignore locked rows.
+ *
+ * Using the id cache, the {@link JdbcChannelMessageStore} will store each
+ * message id in an in-memory collection for the duration of processing. With
+ * that, any polling threads will explicitly exclude those messages from
+ * being polled.
+ *
+ * For this to work, you must setup the corresponding
+ * {@link TransactionSynchronizationFactory}:
+ *
+ *
+ * {@code
+ *
+ *
+ * }
+ *
+ *
+ * This {@link TransactionSynchronizationFactory} is then referenced in the
+ * transaction configuration of the poller:
+ *
+ *
+ * {@code
+ *
+ *
+ * }
+ *
+ *
+ * @param usingIdCache When true the id cache will be used.
+ */
+ public void setUsingIdCache(boolean usingIdCache) {
+ 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> messages = namedParameterJdbcTemplate.query(query, parameters, messageRowMapper);
+
+ Assert.isTrue(messages.size() == 0 || messages.size() == 1);
+ if (messages.size() > 0){
+
+ final Messagemessage = 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 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);
+ }
+
+ /**
+ * 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.
+ * Only applicable if {@link #setUsingIdCache(boolean)} is set to
+ * true
.
+ *
+ * @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);
+ }
+ });
+
+ }
+
+}
diff --git 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
new file mode 100644
index 0000000000..336a22b514
--- /dev/null
+++ 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=?";
+ }
+
+}
diff --git 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
new file mode 100644
index 0000000000..11aea94f45
--- /dev/null
+++ 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();
+
+}
diff --git 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
new file mode 100644
index 0000000000..cd079b661c
--- /dev/null
+++ 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";
+ }
+
+}
diff --git 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
new file mode 100644
index 0000000000..b507a6ab24
--- /dev/null
+++ 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";
+ }
+
+}
diff --git 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
new file mode 100644
index 0000000000..af683fcefe
--- /dev/null
+++ 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> {
+
+ 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"));
+ }
+
+}
diff --git 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
new file mode 100644
index 0000000000..6a021bb35b
--- /dev/null
+++ 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";
+ }
+
+}
diff --git 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
new file mode 100644
index 0000000000..77add08831
--- /dev/null
+++ 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 1.
+ *
+ * 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";
+ }
+
+}
diff --git 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
new file mode 100644
index 0000000000..87d7c9a91a
--- /dev/null
+++ 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";
+ }
+
+}
diff --git 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
new file mode 100644
index 0000000000..b5fbffd699
--- /dev/null
+++ 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;
\ No newline at end of file
diff --git 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
new file mode 100644
index 0000000000..e3498ea73b
--- /dev/null
+++ 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;
\ No newline at end of file
diff --git 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
new file mode 100644
index 0000000000..3e46c287c8
--- /dev/null
+++ 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);
\ No newline at end of file
diff --git 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
new file mode 100644
index 0000000000..ce4d2c90b7
--- /dev/null
+++ 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;
diff --git 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
new file mode 100644
index 0000000000..ce4d2c90b7
--- /dev/null
+++ 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;
diff --git 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
new file mode 100644
index 0000000000..b83c9dce6e
--- /dev/null
+++ 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;
+
diff --git 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
new file mode 100644
index 0000000000..3b9c9caf8f
--- /dev/null
+++ 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;
+
diff --git 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
new file mode 100644
index 0000000000..b83c9dce6e
--- /dev/null
+++ 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;
+
diff --git 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
new file mode 100644
index 0000000000..339f25869a
--- /dev/null
+++ 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);
diff --git 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
new file mode 100644
index 0000000000..8ae9fb85b5
--- /dev/null
+++ 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) ;
\ No newline at end of file
diff --git 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
new file mode 100644
index 0000000000..ad884b801d
--- /dev/null
+++ 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);
diff --git 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
new file mode 100644
index 0000000000..d39aa200b7
--- /dev/null
+++ 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);
\ No newline at end of file
diff --git 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
index 776112989a..8a07c8dc87 100644
--- 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 message = new GenericMessage("foo");
- handler.handleMessage(message);
+ handler.handleMessage(message);
Map 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 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)");
diff --git 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
index 98d53df05a..9864e46d2f 100644
--- 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");
-
+
}
}
diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractJdbcChannelMessageStoreTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractJdbcChannelMessageStoreTests.java
new file mode 100644
index 0000000000..b36d8461d7
--- /dev/null
+++ 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 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));
+ }
+
+}
diff --git 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
new file mode 100644
index 0000000000..86227ad3ad
--- /dev/null
+++ 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()));
+ }
+
+}
diff --git 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
new file mode 100644
index 0000000000..19d689d2d4
--- /dev/null
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/DataSource-common-context.xml
@@ -0,0 +1,16 @@
+
+
+
+
+
+
+
diff --git 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
new file mode 100644
index 0000000000..1fdaabf9ca
--- /dev/null
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/DataSource-derby-context.xml
@@ -0,0 +1,34 @@
+
+
+
+
+
+
+
+
+
+
+
+
diff --git 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
new file mode 100644
index 0000000000..d09f51bf53
--- /dev/null
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/DataSource-hsql-context.xml
@@ -0,0 +1,25 @@
+
+
+
+
+
+
+
+
+
+
diff --git 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
new file mode 100644
index 0000000000..752575bb42
--- /dev/null
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/DataSource-mysql-context.xml
@@ -0,0 +1,22 @@
+
+
+
+
+
+
+
diff --git 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
new file mode 100644
index 0000000000..55fd3f0c58
--- /dev/null
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/DataSource-oracle-context.xml
@@ -0,0 +1,26 @@
+
+
+
+
+
+
+ 3
+ 20
+
+
+
+
+
diff --git 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
new file mode 100644
index 0000000000..1177e4944a
--- /dev/null
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/DataSource-postgres-context.xml
@@ -0,0 +1,23 @@
+
+
+
+
+
+
+
diff --git 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
new file mode 100644
index 0000000000..bad916beec
--- /dev/null
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/DerbyTxTimeoutMessageStoreTests-context.xml
@@ -0,0 +1,20 @@
+
+
+
+
+
diff --git 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
new file mode 100644
index 0000000000..f48528226d
--- /dev/null
+++ 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();
+ }
+
+}
diff --git 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
new file mode 100644
index 0000000000..32fd55b6b1
--- /dev/null
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/HsqlJdbcChannelMessageStoreTests-context.xml
@@ -0,0 +1,11 @@
+
+
+
+
diff --git 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
new file mode 100644
index 0000000000..6e7f2b518b
--- /dev/null
+++ 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();
+ }
+
+}
diff --git 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
new file mode 100644
index 0000000000..8090b96870
--- /dev/null
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/HsqlTxTimeoutMessageStoreTests-context.xml
@@ -0,0 +1,20 @@
+
+
+
+
+
diff --git 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
new file mode 100644
index 0000000000..8b2155470d
--- /dev/null
+++ 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();
+
+ }
+
+}
diff --git 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
new file mode 100644
index 0000000000..b400bfe7ac
--- /dev/null
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/JdbcChannelMessageStoreTests-context.xml
@@ -0,0 +1,8 @@
+
+
+
+
diff --git 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
new file mode 100644
index 0000000000..cbd2a03018
--- /dev/null
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/MySqlJdbcChannelMessageStoreTests-context.xml
@@ -0,0 +1,11 @@
+
+
+
+
diff --git 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
new file mode 100644
index 0000000000..a328780151
--- /dev/null
+++ 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();
+ }
+
+}
diff --git 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
new file mode 100644
index 0000000000..ef71626908
--- /dev/null
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/MySqlTxTimeoutMessageStoreTests-context.xml
@@ -0,0 +1,20 @@
+
+
+
+
+
diff --git 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
new file mode 100644
index 0000000000..b7f816d06f
--- /dev/null
+++ 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();
+ }
+
+}
diff --git 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
new file mode 100644
index 0000000000..c3b4927524
--- /dev/null
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/OracleTxTimeoutMessageStoreTests-context.xml
@@ -0,0 +1,20 @@
+
+
+
+
+
diff --git 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
new file mode 100644
index 0000000000..5b28021488
--- /dev/null
+++ 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();
+ }
+
+}
diff --git 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
new file mode 100644
index 0000000000..b9171cdcd4
--- /dev/null
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/PostgresTxTimeoutMessageStoreTests-context.xml
@@ -0,0 +1,20 @@
+
+
+
+
+
diff --git 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
new file mode 100644
index 0000000000..6bf7b7bf6d
--- /dev/null
+++ 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();
+ }
+
+}
diff --git 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
new file mode 100644
index 0000000000..96c2b16a7c
--- /dev/null
+++ 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 seen = new ConcurrentHashMap();
+
+ 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 getSeenMessages() {
+ return seen;
+ }
+
+ public int getDuplicateMessagesCount() {
+ return duplicateMessagesCount.get();
+ }
+
+}
\ No newline at end of file
diff --git 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
new file mode 100644
index 0000000000..b4f04f17f0
--- /dev/null
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/TxTimeoutMessageStoreTests-context.xml
@@ -0,0 +1,58 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git 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
index d5e8dc87d5..eb53c7bf9f 100644
--- 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
*
*/
diff --git a/spring-integration-jdbc/src/test/resources/log4j.properties b/spring-integration-jdbc/src/test/resources/log4j.properties
index 73cb4696d0..6cb45df171 100644
--- 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
diff --git a/src/reference/docbook/jdbc.xml b/src/reference/docbook/jdbc.xml
index 76908c29f9..196b88889e 100644
--- a/src/reference/docbook/jdbc.xml
+++ b/src/reference/docbook/jdbc.xml
@@ -297,38 +297,149 @@
JDBC Message Store
-
- The JDBC module provides an implementation of the Spring Integration
- MessageStore (important in the Claim Check pattern)
- and MessageGroupStore (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:
-
-
+
+ Spring Integration provides 2 JDBC specifc Message Store implementations.
+ The first one, is the JdbcMessageStore which is
+ suitable to be used in conjunction with Aggregators
+ and the Claimcheck pattern. While it can be used for
+ backing Message Channels as well, you may want to
+ consider using the JdbcChannelMessageStore implementation
+ instead, as it provides a more targeted and scalable implementation.
+
+
+ The Generic JDBC Message Store
+
+ The JDBC module provides an implementation of the Spring Integration
+ MessageStore (important in the Claim Check pattern)
+ and MessageGroupStore (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:
+
]]>
- A JdbcTemplate can be specified instead of a
- DataSource .
-
- Other optional attributes are show in the next example:
-
-
+ A JdbcTemplate can be specified instead of a
+ DataSource .
+
+ Other optional attributes are show in the next example:
+
+ ]]>
+
+ Here we have specified a LobHandler 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_".
+
+
+
+ Backing Message Channels
+
+ If you intent backing Message Channels using JDBC,
+ it is recommended to use the provided JdbcChannelMessageStore
+ implementation instead. It can only be used in conjuntion
+ with Message Channels .
+
+
+ The provided JdbcChannelMessageStore
+ implementation is available since Spring Integration 2.2. .
+
+ Supported Database
+ The JdbcChannelMessageStore uses database specific
+ SQL queries to retrieve messages from the database. Therefore, users must
+ set the ChannelMessageStoreQueryProvider property on the
+ JdbcChannelMessageStore . This
+ channelMessageStoreQueryProvider
+ provides the SQL queries and Spring Integration provides support for
+ the following relational databases:
+
+
+ PostgreSQL
+ HSQLDB
+ MySQL
+ Oracle
+ Derby
+
+
+ If your database is not listed, you can easily extend
+ the AbstractChannelMessageStoreQueryProvider
+ class and provide your own custom queries.
+
+
+
+ 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:
+
+
+ 5 subtle ways you’re using MySQL as a queue, and why it’ll bite you The Database As Queue Anti-Pattern
+
+ Concurrent Polling
+ When polling a Message Channel , you have the option
+ to configure the associated Poller with a
+ TaskExecutor reference.
+
+
+ Keep in mind, though, that if you use a JDBC backed Message Channel 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
+ Multiversion Concurrency Control (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.
+
+
- Here we have specified a LobHandler 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_".
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
Initializing the Database
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
- org.springframework.integration.jdbc package:
- there is a create and a drop script example for a range of common
+ org.springframework.integration.jdbc and in the
+ org.springframework.integration.jdbc.store.channel
+ 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 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 region can be used to keep data
separate for different physical channels that happen to have the same
logical name.
diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml
index 991d35d74e..c44398422d 100644
--- a/src/reference/docbook/whats-new.xml
+++ b/src/reference/docbook/whats-new.xml
@@ -67,7 +67,7 @@
- JDBC Adapter - Stored Procedures Components
+ JDBC Support - Stored Procedures Components
SpEL Support
When using the Stored Procedure components of the Spring Integration
@@ -92,13 +92,22 @@
- JDBC Adapter - Outbound Gateway
+ JDBC Support - Outbound Gateway
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.
+
+ JDBC Support - Channel-specific Message Store Implementation
+
+ A new Message Channel -specific Message Store
+ Implementation has been added, providing a more scalable solution
+ using database-specific SQL queries. For more information please
+ see:
+
File Adapter - Improved File Overwrite/Append Handling