diff --git 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 index ae82a0f55a..3e09169b86 100644 --- 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 @@ -1,11 +1,11 @@ /* - * Copyright 2002-2011 the original author or authors. - * + * 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. @@ -24,19 +24,19 @@ import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.UUID; -import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import javax.sql.DataSource; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + import org.springframework.beans.DirectFieldAccessor; 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.dao.DataAccessException; +import org.springframework.dao.DuplicateKeyException; import org.springframework.integration.Message; import org.springframework.integration.MessageHeaders; import org.springframework.integration.store.AbstractMessageGroupStore; @@ -48,7 +48,6 @@ 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.ResultSetExtractor; import org.springframework.jdbc.core.RowCallbackHandler; import org.springframework.jdbc.core.RowMapper; import org.springframework.jdbc.core.SingleColumnRowMapper; @@ -63,7 +62,7 @@ import org.springframework.util.StringUtils; * Implementation of {@link MessageStore} using a relational database via JDBC. SQL scripts to create the necessary * tables are packaged as org/springframework/integration/jdbc/schema-*.sql, where * is the * target database type. - * + * * @author Dave Syer * @author Oleg Zhurakousky * @since 2.0 @@ -78,8 +77,41 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa */ public static final String DEFAULT_TABLE_PREFIX = "INT_"; + public static final String GROUP_EXISTS = "SELECT COUNT(GROUP_KEY) FROM %PREFIX%MESSAGE_GROUP where GROUP_KEY = ?"; + + private static final String CREATE_MESSAGE_GROUP = "INSERT into %PREFIX%MESSAGE_GROUP" + + "(GROUP_KEY, REGION, MARKED, COMPLETE, LAST_RELEASED_SEQUENCE, CREATED_DATE, UPDATED_DATE)" + + " values (?, ?, 0, 0, 0, ?, ?)"; + private static final String UPDATE_MESSAGE_GROUP = "UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=? where GROUP_KEY=? and REGION=?"; + + private static final String REMOVE_MESSAGE_FROM_GROUP = "DELETE from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY=? and MESSAGE_ID=?"; + + private static final String COUNT_ALL_MESSAGES_IN_GROUPS = "SELECT COUNT(MESSAGE_ID) from %PREFIX%GROUP_TO_MESSAGE"; + + private static final String COUNT_ALL_MESSAGES_IN_GROUP = "SELECT COUNT(MESSAGE_ID) from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY=?"; + + private static final String LIST_MESSAGEIDS_BY_GROUP_KEY = "select MESSAGE_ID, CREATED_DATE " + + "from %PREFIX%MESSAGE where MESSAGE_ID in (select MESSAGE_ID from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY = ?) and REGION=? " + + "ORDER BY CREATED_DATE"; + + private static final String LIST_MESSAGES_BY_GROUP_KEY = "SELECT MESSAGE_ID, MESSAGE_BYTES, CREATED_DATE " + + "from %PREFIX%MESSAGE where MESSAGE_ID in (SELECT MESSAGE_ID from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY = ?) and REGION=? " + + "ORDER BY CREATED_DATE"; + + private static final String POLL_FROM_GROUP = + "SELECT %PREFIX%MESSAGE.MESSAGE_ID, %PREFIX%MESSAGE.MESSAGE_BYTES from %PREFIX%MESSAGE " + + "where %PREFIX%MESSAGE.MESSAGE_ID = " + + "(SELECT min(MESSAGE_ID) from %PREFIX%MESSAGE where CREATED_DATE = " + + "(SELECT min(CREATED_DATE) from %PREFIX%MESSAGE, %PREFIX%GROUP_TO_MESSAGE " + + "where %PREFIX%MESSAGE.MESSAGE_ID = %PREFIX%GROUP_TO_MESSAGE.MESSAGE_ID " + + "and %PREFIX%GROUP_TO_MESSAGE.GROUP_KEY = ? " + + "and %PREFIX%MESSAGE.REGION = ?))"; + + private static final String GET_GROUP_INFO = "SELECT COMPLETE, LAST_RELEASED_SEQUENCE, CREATED_DATE, UPDATED_DATE" + + " from %PREFIX%MESSAGE_GROUP where GROUP_KEY = ?"; + private static final String GET_MESSAGE = "SELECT MESSAGE_ID, CREATED_DATE, MESSAGE_BYTES from %PREFIX%MESSAGE where MESSAGE_ID=? and REGION=?"; - + private static final String GET_GROUP_CREATED_DATE = "SELECT CREATED_DATE from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=?"; private static final String GET_MESSAGE_COUNT = "SELECT COUNT(MESSAGE_ID) from %PREFIX%MESSAGE where REGION=?"; @@ -89,27 +121,18 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa private static final String CREATE_MESSAGE = "INSERT into %PREFIX%MESSAGE(MESSAGE_ID, REGION, CREATED_DATE, MESSAGE_BYTES)" + " values (?, ?, ?, ?)"; - private static final String LIST_MESSAGES_BY_GROUP_KEY = "SELECT MESSAGE_ID, CREATED_DATE, UPDATED_DATE, GROUP_KEY, MESSAGE_BYTES, MARKED, COMPLETE, LAST_RELEASED_SEQUENCE from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=? order by UPDATED_DATE"; - - private static final String LIST_MESSAGEIDS_BY_GROUP_KEY = "SELECT MESSAGE_ID, CREATED_DATE from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=? order by UPDATED_DATE"; - private static final String COUNT_ALL_GROUPS = "SELECT COUNT(GROUP_KEY) from %PREFIX%MESSAGE_GROUP where REGION=?"; - private static final String COUNT_ALL_MESSAGES_IN_GROUP = "SELECT COUNT(MESSAGE_ID) from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? AND REGION=?"; - - private static final String COUNT_ALL_MESSAGES_IN_GROUPS = "SELECT COUNT(MESSAGE_ID) from %PREFIX%MESSAGE_GROUP where REGION=?"; - private static final String COMPLETE_GROUP = "UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=?, COMPLETE=1 where GROUP_KEY=? and REGION=?"; - - private static final String UPDATE_LAST_RELEASED_SEQUENCE = "UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=?, LAST_RELEASED_SEQUENCE=? where GROUP_KEY=? and REGION=?"; - private static final String REMOVE_MESSAGE_FROM_GROUP = "DELETE from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=? and MESSAGE_ID=?"; + private static final String UPDATE_LAST_RELEASED_SEQUENCE = "UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=?, LAST_RELEASED_SEQUENCE=? where GROUP_KEY=? and REGION=?"; private static final String DELETE_MESSAGE_GROUP = "DELETE from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=?"; - private static final String CREATE_MESSAGE_IN_GROUP = "INSERT into %PREFIX%MESSAGE_GROUP(MESSAGE_ID, REGION, CREATED_DATE, UPDATED_DATE, GROUP_KEY, MARKED, COMPLETE, LAST_RELEASED_SEQUENCE)" - + " values (?, ?, ?, ?, ?, 0, 0, 0)"; - + private static final String CREATE_GROUP_TO_MESSAGE = "INSERT into %PREFIX%GROUP_TO_MESSAGE" + + "(GROUP_KEY, MESSAGE_ID)" + + " values (?, ?)"; + private static final String UPDATE_GROUP = "UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=? where GROUP_KEY=? and REGION=?"; private static final String LIST_GROUP_KEYS = "SELECT distinct GROUP_KEY as CREATED from %PREFIX%MESSAGE_GROUP where REGION=?"; @@ -151,7 +174,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa /** * Create a {@link MessageStore} with all mandatory properties. - * + * * @param dataSource a {@link DataSource} */ public JdbcMessageStore(DataSource dataSource) { @@ -159,20 +182,10 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa jdbcTemplate = new JdbcTemplate(dataSource); } - /** - * Replace patterns in the input to produce a valid SQL query. This implementation replaces the table prefix. - * - * @param base the SQL query to be transformed - * @return a transformed query with replacements - */ - protected String getQuery(String base) { - return StringUtils.replace(base, "%PREFIX%", tablePrefix); - } - /** * 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) { @@ -182,7 +195,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa /** * 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 DEFAULT. - * + * * @param region the region name to set */ public void setRegion(String region) { @@ -192,7 +205,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa /** * The JDBC {@link DataSource} to use when interacting with the database. Either this property can be set or the * {@link #setJdbcTemplate(JdbcOperations) jdbcTemplate}. - * + * * @param dataSource a {@link DataSource} */ public void setDataSource(DataSource dataSource) { @@ -202,7 +215,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa /** * The {@link JdbcOperations} to use when interacting with the database. Either this property can be set or the * {@link #setDataSource(DataSource) dataSource}. - * + * * @param jdbcTemplate a {@link JdbcOperations} */ public void setJdbcTemplate(JdbcOperations jdbcTemplate) { @@ -212,7 +225,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa /** * 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) { @@ -221,7 +234,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa /** * A converter for serializing messages to byte arrays for storage. - * + * * @param serializer the serializer to set */ @SuppressWarnings("unchecked") @@ -231,7 +244,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa /** * A converter for deserializing byte arrays to messages. - * + * * @param deserializer the deserializer to set */ @SuppressWarnings({ "unchecked", "rawtypes" }) @@ -241,7 +254,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa /** * Check mandatory properties (data source and incrementer). - * + * * @throws Exception */ public void afterPropertiesSet() throws Exception { @@ -288,17 +301,19 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa final long createdDate = System.currentTimeMillis(); Message result = MessageBuilder.fromMessage(message).setHeader(SAVED_KEY, Boolean.TRUE) .setHeader(CREATED_DATE_KEY, new Long(createdDate)).build(); - + 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(CREATE_MESSAGE), new PreparedStatementSetter() { public void setValues(PreparedStatement ps) throws SQLException { - logger.debug("Inserting message with id key=" + messageId); + if (logger.isDebugEnabled()){ + logger.debug("Inserting message with id key=" + messageId); + } ps.setString(1, messageId); ps.setString(2, region); ps.setTimestamp(3, new Timestamp(createdDate)); @@ -310,92 +325,97 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa public MessageGroup addMessageToGroup(Object groupId, Message message) { final String groupKey = getKey(groupId); - final long updatedDate = System.currentTimeMillis(); - final long createdDate = this.getGroupCreatedDate(groupKey); - final String messageId = getKey(message.getHeaders().getId()); - + boolean groupNotExist = jdbcTemplate.queryForInt(this.getQuery(GROUP_EXISTS), groupKey) < 1; - jdbcTemplate.update(getQuery(CREATE_MESSAGE_IN_GROUP), new PreparedStatementSetter() { + final Timestamp updatedDate = new Timestamp(System.currentTimeMillis()); + + final Timestamp createdDate = groupNotExist ? + updatedDate : + jdbcTemplate.queryForObject(getQuery(GET_GROUP_CREATED_DATE), new Object[] { groupKey, region}, Timestamp.class); + + if (groupNotExist){ + try { + this.doCreateMessageGroup(groupKey, createdDate); + } catch (DuplicateKeyException e) { + logger.warn("Lost race to create group; attempting update instead", e); + this.doUpdateMessageGroup(groupKey, updatedDate); + } + } + else { + this.doUpdateMessageGroup(groupKey, updatedDate); + } + + this.addMessage(message); + + jdbcTemplate.update(getQuery(CREATE_GROUP_TO_MESSAGE), new PreparedStatementSetter() { public void setValues(PreparedStatement ps) throws SQLException { - logger.debug("Inserting message with id key=" + messageId + " and created date=" + createdDate); - ps.setString(1, messageId); - ps.setString(2, region); - if (createdDate == 0){ - ps.setTimestamp(3, new Timestamp(updatedDate)); + if (logger.isDebugEnabled()){ + logger.debug("Inserting message with id key=" + messageId + " and created date=" + createdDate); } - else { - ps.setTimestamp(3, new Timestamp(createdDate)); - } - - ps.setTimestamp(4, new Timestamp(updatedDate)); - ps.setString(5, groupKey); + ps.setString(1, groupKey); + ps.setString(2, messageId); } }); - this.addMessage(message); return getMessageGroup(groupId); } + @Override @ManagedAttribute public int getMessageGroupCount() { return jdbcTemplate.queryForInt(getQuery(COUNT_ALL_GROUPS), region); } + @Override @ManagedAttribute public int getMessageCountForAllMessageGroups() { - return jdbcTemplate.queryForInt(getQuery(COUNT_ALL_MESSAGES_IN_GROUPS), region); + return jdbcTemplate.queryForInt(getQuery(COUNT_ALL_MESSAGES_IN_GROUPS)); } @ManagedAttribute public int messageGroupSize(Object groupId) { String key = getKey(groupId); - return jdbcTemplate.queryForInt(getQuery(COUNT_ALL_MESSAGES_IN_GROUP), key, region); + return jdbcTemplate.queryForInt(getQuery(COUNT_ALL_MESSAGES_IN_GROUP), key); } public MessageGroup getMessageGroup(Object groupId) { String key = getKey(groupId); - final List> messages = new ArrayList>(); - final AtomicReference date = new AtomicReference(); + final AtomicReference createDate = new AtomicReference(); final AtomicReference updateDate = new AtomicReference(); final AtomicReference completeFlag = new AtomicReference(); final AtomicReference lastReleasedSequenceRef = new AtomicReference(); - - final AtomicInteger size = new AtomicInteger(); - jdbcTemplate.query(getQuery(LIST_MESSAGES_BY_GROUP_KEY), new Object[] { key, region }, - + + List> messages = jdbcTemplate.query(getQuery(LIST_MESSAGES_BY_GROUP_KEY), new Object[] { key, region }, mapper); + + jdbcTemplate.query(getQuery(GET_GROUP_INFO), new Object[] { key}, new RowCallbackHandler() { public void processRow(ResultSet rs) throws SQLException { - size.incrementAndGet(); - - messages.add(getMessage(UUID.fromString(rs.getString("MESSAGE_ID")))); - - date.set(rs.getTimestamp("CREATED_DATE")); - updateDate.set(rs.getTimestamp("UPDATED_DATE")); - + + createDate.set(rs.getTimestamp("CREATED_DATE")); + completeFlag.set(rs.getInt("COMPLETE") > 0); - + lastReleasedSequenceRef.set(rs.getInt("LAST_RELEASED_SEQUENCE")); } }); - - if (size.get() == 0){ + + if (messages.size() == 0){ return new SimpleMessageGroup(groupId); } - Assert.state(date.get() != null, "Could not locate created date for groupId=" + groupId); + Assert.state(createDate.get() != null, "Could not locate created date for groupId=" + groupId); Assert.state(updateDate.get() != null, "Could not locate updated date for groupId=" + groupId); - long timestamp = date.get().getTime(); + + long timestamp = createDate.get().getTime(); boolean complete = completeFlag.get().booleanValue(); + SimpleMessageGroup messageGroup = new SimpleMessageGroup(messages, groupId, timestamp, complete); - if (updateDate.get() != null){ - messageGroup.setLastModified(updateDate.get().getTime()); - } + messageGroup.setLastModified(updateDate.get().getTime()); + int lastReleasedSequenceNumber = lastReleasedSequenceRef.get(); - if (lastReleasedSequenceNumber > 0){ - messageGroup.setLastReleasedMessageSequenceNumber(lastReleasedSequenceNumber); - } - + messageGroup.setLastReleasedMessageSequenceNumber(lastReleasedSequenceNumber); + return messageGroup; } @@ -405,10 +425,11 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa jdbcTemplate.update(getQuery(REMOVE_MESSAGE_FROM_GROUP), new PreparedStatementSetter() { public void setValues(PreparedStatement ps) throws SQLException { - logger.debug("Removing message from group with group key=" + groupKey); + if (logger.isDebugEnabled()){ + logger.debug("Removing message from group with group key=" + groupKey); + } ps.setString(1, groupKey); - ps.setString(2, region); - ps.setString(3, messageId); + ps.setString(2, messageId); } }); this.removeMessage(messageToRemove.getHeaders().getId()); @@ -419,28 +440,31 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa public void removeMessageGroup(Object groupId) { final String groupKey = getKey(groupId); - + for (UUID messageIds : this.getMessageIdsForGroup(groupId)) { this.removeMessage(messageIds); } jdbcTemplate.update(getQuery(DELETE_MESSAGE_GROUP), new PreparedStatementSetter() { public void setValues(PreparedStatement ps) throws SQLException { - logger.debug("Marking messages with group key=" + groupKey); + if (logger.isDebugEnabled()){ + logger.debug("Marking messages with group key=" + groupKey); + } ps.setString(1, groupKey); ps.setString(2, region); } }); - } - + public void completeGroup(Object groupId) { final long updatedDate = System.currentTimeMillis(); final String groupKey = getKey(groupId); - + jdbcTemplate.update(getQuery(COMPLETE_GROUP), new PreparedStatementSetter() { public void setValues(PreparedStatement ps) throws SQLException { - logger.debug("Completing MessageGroup: " + groupKey); + if (logger.isDebugEnabled()){ + logger.debug("Completing MessageGroup: " + groupKey); + } ps.setTimestamp(1, new Timestamp(updatedDate)); ps.setString(2, groupKey); ps.setString(3, region); @@ -452,10 +476,12 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa Assert.notNull(groupId, "'groupId' must not be null"); final long updatedDate = System.currentTimeMillis(); final String groupKey = getKey(groupId); - + jdbcTemplate.update(getQuery(UPDATE_LAST_RELEASED_SEQUENCE), new PreparedStatementSetter() { public void setValues(PreparedStatement ps) throws SQLException { - logger.debug("Updating the sequence number of the last released Message in the MessageGroup: " + groupKey); + if (logger.isDebugEnabled()){ + logger.debug("Updating the sequence number of the last released Message in the MessageGroup: " + groupKey); + } ps.setTimestamp(1, new Timestamp(updatedDate)); ps.setInt(2, sequenceNumber); ps.setString(3, groupKey); @@ -464,31 +490,17 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa }); this.updateMessageGroup(groupKey); } - - public Message pollMessageFromGroup(final Object groupId) { + + public Message pollMessageFromGroup(Object groupId) { String key = getKey(groupId); - - Message message = jdbcTemplate.query(getQuery(LIST_MESSAGEIDS_BY_GROUP_KEY), new Object[] { key, region }, - new ResultSetExtractor>() { - public Message extractData(ResultSet rs) - throws SQLException, DataAccessException { - while (rs.next()) { - UUID uuid = UUID.fromString(rs.getString(1)); - if (uuid != null){ - Message message = getMessage(uuid); - if (message != null){ - removeMessageFromGroup(groupId, message); - return message; - } - } - } - return null; - } - }); - this.updateMessageGroup(key); - return message; + + Message polledMessage = this.doPollForMessage(key); + if (polledMessage != null){ + this.removeMessageFromGroup(groupId, polledMessage); + } + return polledMessage; } - + public Iterator iterator() { final Iterator iterator = jdbcTemplate.query(getQuery(LIST_GROUP_KEYS), new Object[] { region }, @@ -507,15 +519,78 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa public void remove() { throw new UnsupportedOperationException("Cannot remove MessageGroup from this iterator."); } - }; - } - + + /** + * Replace patterns in the input to produce a valid SQL query. This implementation replaces the table prefix. + * + * @param base the SQL query to be transformed + * @return a transformed query with replacements + */ + protected String getQuery(String base) { + return StringUtils.replace(base, "%PREFIX%", tablePrefix); + } + + /** + * To be used to get a reference to JdbcOperations + * in case this class is subclassed + * + * @return + */ + protected JdbcOperations getJdbcOperations(){ + return this.jdbcTemplate; + } + + /** + * This method executes a call to the DB to get the oldest Message in the MessageGroup + * Override this method if need to. For example if you DB supports advanced function such as FIRST etc. + * + * @param String representation of message group ID + * @return could be null if query produced no Messages + */ + protected Message doPollForMessage(String groupIdKey) { + List> messages = jdbcTemplate.query(getQuery(POLL_FROM_GROUP), new Object[] { groupIdKey, region }, mapper); + Assert.isTrue(messages.size() == 0 || messages.size() == 1); + if (messages.size() > 0){ + return messages.get(0); + } + return null; + } + + private void doCreateMessageGroup(final String groupKey, final Timestamp createdDate){ + jdbcTemplate.update(getQuery(CREATE_MESSAGE_GROUP), new PreparedStatementSetter() { + public void setValues(PreparedStatement ps) throws SQLException { + if (logger.isDebugEnabled()){ + logger.debug("Creating message group with id key=" + groupKey + " and created date=" + createdDate); + } + ps.setString(1, groupKey); + ps.setString(2, region); + ps.setTimestamp(3, createdDate); + ps.setTimestamp(4, createdDate); + } + }); + } + + private void doUpdateMessageGroup(final String groupKey, final Timestamp updatedDate){ + jdbcTemplate.update(getQuery(UPDATE_MESSAGE_GROUP), new PreparedStatementSetter() { + public void setValues(PreparedStatement ps) throws SQLException { + if (logger.isDebugEnabled()){ + logger.debug("Updating message group with id key=" + groupKey + " and updated date=" + updatedDate); + } + ps.setTimestamp(1, updatedDate); + ps.setString(2, groupKey); + ps.setString(3, region); + } + }); + } + private void updateMessageGroup(final String groupId){ jdbcTemplate.update(getQuery(UPDATE_GROUP), new PreparedStatementSetter() { public void setValues(PreparedStatement ps) throws SQLException { - logger.debug("Updating MessageGroup: " + groupId); + if (logger.isDebugEnabled()){ + logger.debug("Updating MessageGroup: " + groupId); + } ps.setTimestamp(1, new Timestamp(System.currentTimeMillis())); ps.setString(2, groupId); ps.setString(3, region); @@ -525,49 +600,28 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa private List getMessageIdsForGroup(Object groupId){ String key = getKey(groupId); - + final List messageIds = new ArrayList(); - + jdbcTemplate.query(getQuery(LIST_MESSAGEIDS_BY_GROUP_KEY), new Object[] { key, region }, new RowCallbackHandler() { public void processRow(ResultSet rs) throws SQLException { messageIds.add(UUID.fromString(rs.getString(1))); } - } ); return messageIds; } - - private String getKey(Object input) { return input == null ? null : UUIDConverter.getUUID(input).toString(); } - - private long getGroupCreatedDate(String groupKey) { - final AtomicReference date = new AtomicReference(); - this.jdbcTemplate.query(getQuery(GET_GROUP_CREATED_DATE), new Object[] { groupKey, region }, - - new RowCallbackHandler() { - public void processRow(ResultSet rs) throws SQLException { - date.set(rs.getTimestamp("CREATED_DATE").getTime()); - } - }); - Long returnedDate = date.get(); - if (returnedDate == null){ - return 0; - } - else { - return returnedDate; - } - } /** * Convenience class to be used to unpack a 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 Dave Syer */ private class MessageMapper implements RowMapper> { diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql index 0aa02f9f3c..a06f3b89d2 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql @@ -7,15 +7,20 @@ CREATE TABLE INT_MESSAGE ( MESSAGE_BYTES BLOB ); -CREATE TABLE INT_MESSAGE_GROUP ( - MESSAGE_ID CHAR(36) NOT NULL, +CREATE INDEX INT_MESSAGE_IX1 ON INT_MESSAGE (CREATED_DATE); + +CREATE TABLE INT_GROUP_TO_MESSAGE ( GROUP_KEY CHAR(36) NOT NULL, + MESSAGE_ID CHAR(36) NOT NULL, + constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, MESSAGE_ID) +); + +CREATE TABLE INT_MESSAGE_GROUP ( + GROUP_KEY CHAR(36) NOT NULL PRIMARY KEY, REGION VARCHAR(100), MARKED BIGINT, COMPLETE BIGINT, LAST_RELEASED_SEQUENCE BIGINT, CREATED_DATE TIMESTAMP NOT NULL, - UPDATED_DATE TIMESTAMP DEFAULT NULL, - MESSAGE_BYTES BLOB, - constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, GROUP_KEY) -); + UPDATED_DATE TIMESTAMP DEFAULT NULL +); \ No newline at end of file diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql index 0aa02f9f3c..a06f3b89d2 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql @@ -7,15 +7,20 @@ CREATE TABLE INT_MESSAGE ( MESSAGE_BYTES BLOB ); -CREATE TABLE INT_MESSAGE_GROUP ( - MESSAGE_ID CHAR(36) NOT NULL, +CREATE INDEX INT_MESSAGE_IX1 ON INT_MESSAGE (CREATED_DATE); + +CREATE TABLE INT_GROUP_TO_MESSAGE ( GROUP_KEY CHAR(36) NOT NULL, + MESSAGE_ID CHAR(36) NOT NULL, + constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, MESSAGE_ID) +); + +CREATE TABLE INT_MESSAGE_GROUP ( + GROUP_KEY CHAR(36) NOT NULL PRIMARY KEY, REGION VARCHAR(100), MARKED BIGINT, COMPLETE BIGINT, LAST_RELEASED_SEQUENCE BIGINT, CREATED_DATE TIMESTAMP NOT NULL, - UPDATED_DATE TIMESTAMP DEFAULT NULL, - MESSAGE_BYTES BLOB, - constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, GROUP_KEY) -); + UPDATED_DATE TIMESTAMP DEFAULT NULL +); \ No newline at end of file diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-db2.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-db2.sql index 45929cec1a..77cc319e83 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-db2.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-db2.sql @@ -2,3 +2,5 @@ DROP TABLE INT_MESSAGE ; DROP TABLE INT_MESSAGE_GROUP ; +DROP TABLE INT_GROUP_TO_MESSAGE ; +DROP INDEX INT_MESSAGE_IX1 ; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-derby.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-derby.sql index 45929cec1a..77cc319e83 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-derby.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-derby.sql @@ -2,3 +2,5 @@ DROP TABLE INT_MESSAGE ; DROP TABLE INT_MESSAGE_GROUP ; +DROP TABLE INT_GROUP_TO_MESSAGE ; +DROP INDEX INT_MESSAGE_IX1 ; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-h2.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-h2.sql index 35ff5cb399..dd4c973773 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-h2.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-h2.sql @@ -2,3 +2,5 @@ DROP TABLE INT_MESSAGE IF EXISTS; DROP TABLE INT_MESSAGE_GROUP IF EXISTS; +DROP TABLE INT_GROUP_TO_MESSAGE IF EXISTS; +DROP INDEX INT_MESSAGE_IX1 IF EXISTS; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-hsqldb.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-hsqldb.sql index 35ff5cb399..dd4c973773 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-hsqldb.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-hsqldb.sql @@ -2,3 +2,5 @@ DROP TABLE INT_MESSAGE IF EXISTS; DROP TABLE INT_MESSAGE_GROUP IF EXISTS; +DROP TABLE INT_GROUP_TO_MESSAGE IF EXISTS; +DROP INDEX INT_MESSAGE_IX1 IF EXISTS; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-mysql.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-mysql.sql index 33904f1159..fce4d8c950 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-mysql.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-mysql.sql @@ -2,3 +2,5 @@ DROP TABLE IF EXISTS INT_MESSAGE ; DROP TABLE IF EXISTS INT_MESSAGE_GROUP ; +DROP TABLE IF EXISTS INT_GROUP_TO_MESSAGE ; +DROP INDEX IF EXISTS INT_MESSAGE_IX1 ; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-oracle10g.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-oracle10g.sql index 45929cec1a..77cc319e83 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-oracle10g.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-oracle10g.sql @@ -2,3 +2,5 @@ DROP TABLE INT_MESSAGE ; DROP TABLE INT_MESSAGE_GROUP ; +DROP TABLE INT_GROUP_TO_MESSAGE ; +DROP INDEX INT_MESSAGE_IX1 ; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-postgresql.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-postgresql.sql index 45929cec1a..77cc319e83 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-postgresql.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-postgresql.sql @@ -2,3 +2,5 @@ DROP TABLE INT_MESSAGE ; DROP TABLE INT_MESSAGE_GROUP ; +DROP TABLE INT_GROUP_TO_MESSAGE ; +DROP INDEX INT_MESSAGE_IX1 ; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sqlserver.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sqlserver.sql index 45929cec1a..77cc319e83 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sqlserver.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sqlserver.sql @@ -2,3 +2,5 @@ DROP TABLE INT_MESSAGE ; DROP TABLE INT_MESSAGE_GROUP ; +DROP TABLE INT_GROUP_TO_MESSAGE ; +DROP INDEX INT_MESSAGE_IX1 ; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sybase.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sybase.sql index 45929cec1a..77cc319e83 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sybase.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sybase.sql @@ -2,3 +2,5 @@ DROP TABLE INT_MESSAGE ; DROP TABLE INT_MESSAGE_GROUP ; +DROP TABLE INT_GROUP_TO_MESSAGE ; +DROP INDEX INT_MESSAGE_IX1 ; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql index 046a436c95..02b300b2f7 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql @@ -7,15 +7,20 @@ CREATE TABLE INT_MESSAGE ( MESSAGE_BYTES LONGVARBINARY ); -CREATE TABLE INT_MESSAGE_GROUP ( - MESSAGE_ID CHAR(36) NOT NULL, +CREATE INDEX INT_MESSAGE_IX1 ON INT_MESSAGE (CREATED_DATE); + +CREATE TABLE INT_GROUP_TO_MESSAGE ( GROUP_KEY CHAR(36) NOT NULL, + MESSAGE_ID CHAR(36) NOT NULL, + constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, MESSAGE_ID) +); + +CREATE TABLE INT_MESSAGE_GROUP ( + GROUP_KEY CHAR(36) NOT NULL PRIMARY KEY, REGION VARCHAR(100), MARKED BIGINT, COMPLETE BIGINT, LAST_RELEASED_SEQUENCE BIGINT, CREATED_DATE TIMESTAMP NOT NULL, - UPDATED_DATE TIMESTAMP DEFAULT NULL, - MESSAGE_BYTES LONGVARBINARY, - constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, GROUP_KEY) -); + UPDATED_DATE TIMESTAMP DEFAULT NULL +); \ No newline at end of file diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql index 046a436c95..02b300b2f7 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql @@ -7,15 +7,20 @@ CREATE TABLE INT_MESSAGE ( MESSAGE_BYTES LONGVARBINARY ); -CREATE TABLE INT_MESSAGE_GROUP ( - MESSAGE_ID CHAR(36) NOT NULL, +CREATE INDEX INT_MESSAGE_IX1 ON INT_MESSAGE (CREATED_DATE); + +CREATE TABLE INT_GROUP_TO_MESSAGE ( GROUP_KEY CHAR(36) NOT NULL, + MESSAGE_ID CHAR(36) NOT NULL, + constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, MESSAGE_ID) +); + +CREATE TABLE INT_MESSAGE_GROUP ( + GROUP_KEY CHAR(36) NOT NULL PRIMARY KEY, REGION VARCHAR(100), MARKED BIGINT, COMPLETE BIGINT, LAST_RELEASED_SEQUENCE BIGINT, CREATED_DATE TIMESTAMP NOT NULL, - UPDATED_DATE TIMESTAMP DEFAULT NULL, - MESSAGE_BYTES LONGVARBINARY, - constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, GROUP_KEY) -); + UPDATED_DATE TIMESTAMP DEFAULT NULL +); \ No newline at end of file diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql index f9e3062ccd..faffeb2208 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql @@ -7,15 +7,20 @@ CREATE TABLE INT_MESSAGE ( MESSAGE_BYTES BLOB ) ENGINE=InnoDB; -CREATE TABLE INT_MESSAGE_GROUP ( - MESSAGE_ID CHAR(36) NOT NULL, +CREATE INDEX INT_MESSAGE_IX1 ON INT_MESSAGE (CREATED_DATE) ENGINE=InnoDB; + +CREATE TABLE INT_GROUP_TO_MESSAGE ( GROUP_KEY CHAR(36) NOT NULL, + MESSAGE_ID CHAR(36) NOT NULL, + constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, MESSAGE_ID) +) ENGINE=InnoDB; + +CREATE TABLE INT_MESSAGE_GROUP ( + GROUP_KEY CHAR(36) NOT NULL PRIMARY KEY, REGION VARCHAR(100), MARKED BIGINT, COMPLETE BIGINT, LAST_RELEASED_SEQUENCE BIGINT, CREATED_DATE DATETIME NOT NULL, - UPDATED_DATE DATETIME DEFAULT NULL, - MESSAGE_BYTES BLOB, - constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, GROUP_KEY) -) ENGINE=InnoDB; + UPDATED_DATE DATETIME DEFAULT NULL +) ENGINE=InnoDB; \ No newline at end of file diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle10g.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle10g.sql index 6937ed165e..c0f7ff2180 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle10g.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle10g.sql @@ -7,15 +7,20 @@ CREATE TABLE INT_MESSAGE ( MESSAGE_BYTES BLOB ); -CREATE TABLE INT_MESSAGE_GROUP ( - MESSAGE_ID CHAR(36) NOT NULL, +CREATE INDEX INT_MESSAGE_IX1 ON INT_MESSAGE (CREATED_DATE); + +CREATE TABLE INT_GROUP_TO_MESSAGE ( GROUP_KEY CHAR(36) NOT NULL, + MESSAGE_ID CHAR(36) NOT NULL, + constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, MESSAGE_ID) +); + +CREATE TABLE INT_MESSAGE_GROUP ( + GROUP_KEY CHAR(36) NOT NULL PRIMARY KEY, REGION VARCHAR2(100), MARKED NUMBER(19,0), COMPLETE NUMBER(19,0), LAST_RELEASED_SEQUENCE NUMBER(19,0), CREATED_DATE TIMESTAMP NOT NULL, - UPDATED_DATE TIMESTAMP DEFAULT NULL, - MESSAGE_BYTES BLOB, - constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, GROUP_KEY) -); + UPDATED_DATE TIMESTAMP DEFAULT NULL +); \ No newline at end of file diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql index 3e7917e848..47aa0553e5 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql @@ -7,15 +7,20 @@ CREATE TABLE INT_MESSAGE ( MESSAGE_BYTES BYTEA ); -CREATE TABLE INT_MESSAGE_GROUP ( - MESSAGE_ID CHAR(36) NOT NULL, +CREATE INDEX INT_MESSAGE_IX1 ON INT_MESSAGE (CREATED_DATE); + +CREATE TABLE INT_GROUP_TO_MESSAGE ( GROUP_KEY CHAR(36) NOT NULL, + MESSAGE_ID CHAR(36) NOT NULL, + constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, MESSAGE_ID) +); + +CREATE TABLE INT_MESSAGE_GROUP ( + GROUP_KEY CHAR(36) NOT NULL PRIMARY KEY, REGION VARCHAR(100), MARKED BIGINT, COMPLETE BIGINT, LAST_RELEASED_SEQUENCE BIGINT, CREATED_DATE TIMESTAMP NOT NULL, - UPDATED_DATE TIMESTAMP DEFAULT NULL, - MESSAGE_BYTES BYTEA, - constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, GROUP_KEY) -); + UPDATED_DATE TIMESTAMP DEFAULT NULL +); \ No newline at end of file diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql index 73b3e8195c..1f229e897e 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql @@ -7,15 +7,20 @@ CREATE TABLE INT_MESSAGE ( MESSAGE_BYTES IMAGE ); -CREATE TABLE INT_MESSAGE_GROUP ( - MESSAGE_ID CHAR(36) NOT NULL, +CREATE INDEX INT_MESSAGE_IX1 ON INT_MESSAGE (CREATED_DATE); + +CREATE TABLE INT_GROUP_TO_MESSAGE ( GROUP_KEY CHAR(36) NOT NULL, + MESSAGE_ID CHAR(36) NOT NULL, + constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, MESSAGE_ID) +); + +CREATE TABLE INT_MESSAGE_GROUP ( + GROUP_KEY CHAR(36) NOT NULL PRIMARY KEY, REGION VARCHAR(100), MARKED BIGINT, COMPLETE BIGINT, LAST_RELEASED_SEQUENCE BIGINT, CREATED_DATE DATETIME NOT NULL, - UPDATED_DATE DATETIME DEFAULT NULL, - MESSAGE_BYTES IMAGE, - constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, GROUP_KEY) -); + UPDATED_DATE DATETIME DEFAULT NULL +); \ No newline at end of file diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql index 4d12b8ed39..3c1d7c2193 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql @@ -5,17 +5,22 @@ CREATE TABLE INT_MESSAGE ( REGION VARCHAR(100), CREATED_DATE DATETIME NOT NULL, MESSAGE_BYTES IMAGE -) LOCK DATAROWS WITH EXP_ROW_SIZE=1; +) LOCK DATAROWS; + +CREATE INDEX INT_MESSAGE_IX1 ON INT_MESSAGE (CREATED_DATE) LOCK DATAROWS; + +CREATE TABLE INT_GROUP_TO_MESSAGE ( + GROUP_KEY CHAR(36) NOT NULL, + MESSAGE_ID CHAR(36) NOT NULL, + constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, MESSAGE_ID) +) LOCK DATAROWS; CREATE TABLE INT_MESSAGE_GROUP ( - MESSAGE_ID CHAR(36) NOT NULL, - GROUP_KEY CHAR(36) NOT NULL, + GROUP_KEY CHAR(36) NOT NULL PRIMARY KEY, REGION VARCHAR(100), MARKED BIGINT, COMPLETE BIGINT, LAST_RELEASED_SEQUENCE BIGINT, CREATED_DATE DATETIME NOT NULL, - UPDATED_DATE DATETIME DEFAULT NULL, - MESSAGE_BYTES IMAGE, - constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, GROUP_KEY) -) LOCK DATAROWS WITH EXP_ROW_SIZE=1; + UPDATED_DATE DATETIME DEFAULT NULL +) LOCK DATAROWS; \ No newline at end of file diff --git a/spring-integration-jdbc/src/main/sql/destroy.sql.vpp b/spring-integration-jdbc/src/main/sql/destroy.sql.vpp index e23e8bc539..ed223b3683 100644 --- a/spring-integration-jdbc/src/main/sql/destroy.sql.vpp +++ b/spring-integration-jdbc/src/main/sql/destroy.sql.vpp @@ -1,2 +1,4 @@ DROP TABLE $!{IFEXISTSBEFORE} INT_MESSAGE $!{IFEXISTS}; DROP TABLE $!{IFEXISTSBEFORE} INT_MESSAGE_GROUP $!{IFEXISTS}; +DROP TABLE $!{IFEXISTSBEFORE} INT_GROUP_TO_MESSAGE $!{IFEXISTS}; +DROP INDEX $!{IFEXISTSBEFORE} INT_MESSAGE_IX1 $!{IFEXISTS}; diff --git a/spring-integration-jdbc/src/main/sql/schema.sql.vpp b/spring-integration-jdbc/src/main/sql/schema.sql.vpp index 0ffb04b69d..9a0ebb7684 100644 --- a/spring-integration-jdbc/src/main/sql/schema.sql.vpp +++ b/spring-integration-jdbc/src/main/sql/schema.sql.vpp @@ -7,15 +7,20 @@ CREATE TABLE INT_MESSAGE ( MESSAGE_BYTES ${BLOB} )#if(${VOODOO}) ${VOODOO}#end; -CREATE TABLE INT_MESSAGE_GROUP ( - MESSAGE_ID CHAR(36) NOT NULL, +CREATE INDEX INT_MESSAGE_IX1 ON INT_MESSAGE (CREATED_DATE)#if(${VOODOO}) ${VOODOO}#end; + +CREATE TABLE INT_GROUP_TO_MESSAGE ( GROUP_KEY CHAR(36) NOT NULL, + MESSAGE_ID CHAR(36) NOT NULL, + constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, MESSAGE_ID) +)#if(${VOODOO}) ${VOODOO}#end; + +CREATE TABLE INT_MESSAGE_GROUP ( + GROUP_KEY CHAR(36) NOT NULL PRIMARY KEY, REGION ${VARCHAR}(100), MARKED ${BIGINT}, COMPLETE ${BIGINT}, LAST_RELEASED_SEQUENCE ${BIGINT}, CREATED_DATE ${TIMESTAMP} NOT NULL, - UPDATED_DATE ${TIMESTAMP} DEFAULT NULL, - MESSAGE_BYTES ${BLOB}, - constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, GROUP_KEY) -)#if(${VOODOO}) ${VOODOO}#end; + UPDATED_DATE ${TIMESTAMP} DEFAULT NULL +)#if(${VOODOO}) ${VOODOO}#end; \ No newline at end of file diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java index ad9b6220e3..f06557f91c 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * 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. @@ -282,22 +282,6 @@ public class JdbcMessageStoreTests { assertEquals("bar", messageStore.pollMessageFromGroup(groupId).getPayload()); } - @Test - @Transactional - public void testExpireMessageGroup() throws Exception { - String groupId = "X"; - Message message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build(); - messageStore.addMessageToGroup(groupId, message); - messageStore.registerMessageGroupExpiryCallback(new MessageGroupCallback() { - public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { - messageGroupStore.removeMessageGroup(group.getGroupId()); - } - }); - messageStore.expireMessageGroups(-10000); - MessageGroup group = messageStore.getMessageGroup(groupId); - assertEquals(0, group.size()); - } - @Test @Transactional public void testExpireMessageGroupOnCreateOnly() throws Exception { @@ -313,8 +297,8 @@ public class JdbcMessageStoreTests { messageStore.expireMessageGroups(2000); MessageGroup group = messageStore.getMessageGroup(groupId); assertEquals(1, group.size()); - Thread.sleep(2000); messageStore.addMessageToGroup(groupId, MessageBuilder.withPayload("bar").setCorrelationId(groupId).build()); + Thread.sleep(2001); messageStore.expireMessageGroups(2000); group = messageStore.getMessageGroup(groupId); assertEquals(0, group.size()); @@ -338,13 +322,40 @@ public class JdbcMessageStoreTests { assertEquals(1, group.size()); Thread.sleep(2000); messageStore.addMessageToGroup(groupId, MessageBuilder.withPayload("bar").setCorrelationId(groupId).build()); - messageStore.expireMessageGroups(2000); group = messageStore.getMessageGroup(groupId); assertEquals(2, group.size()); Thread.sleep(2000); - messageStore.expireMessageGroups(1000); + messageStore.expireMessageGroups(2000); group = messageStore.getMessageGroup(groupId); assertEquals(0, group.size()); } + @Test + @Transactional + public void testMessagePollingFromTheGroup() throws Exception { + String groupId = "X"; + Message message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build(); + messageStore.addMessageToGroup(groupId, message); + + messageStore.addMessageToGroup(groupId, MessageBuilder.withPayload("bar").setCorrelationId(groupId).build()); + messageStore.addMessageToGroup(groupId, MessageBuilder.withPayload("baz").setCorrelationId(groupId).build()); + messageStore.addMessageToGroup("Y", MessageBuilder.withPayload("barA").setCorrelationId(groupId).build()); + messageStore.addMessageToGroup("Y", MessageBuilder.withPayload("bazA").setCorrelationId(groupId).build()); + MessageGroup group = messageStore.getMessageGroup("X"); + assertEquals(3, group.size()); + + Message message1 = messageStore.pollMessageFromGroup("X"); + assertNotNull(message1); + assertEquals("foo", message1.getPayload()); + System.out.println("Polled Message" + message1); + group = messageStore.getMessageGroup("X"); + assertEquals(2, group.size()); + + Message message2 = messageStore.pollMessageFromGroup("X"); + assertNotNull(message2); + assertEquals("bar", message2.getPayload()); + group = messageStore.getMessageGroup("X"); + assertEquals(1, group.size()); + } + } diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcOutboundChannelAdapterWithinChainTests-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcOutboundChannelAdapterWithinChainTests-context.xml index cf586e4aee..7df0ad45b2 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcOutboundChannelAdapterWithinChainTests-context.xml +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcOutboundChannelAdapterWithinChainTests-context.xml @@ -1,17 +1,20 @@ + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" + xmlns:int="http://www.springframework.org/schema/integration" + xmlns:jdbc="http://www.springframework.org/schema/jdbc" + 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/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"> - - - + + + + + + diff --git a/spring-integration-jdbc/src/test/resources/drop-derby-stored-procedures.sql b/spring-integration-jdbc/src/test/resources/drop-derby-stored-procedures.sql new file mode 100644 index 0000000000..a6a4c5db96 --- /dev/null +++ b/spring-integration-jdbc/src/test/resources/drop-derby-stored-procedures.sql @@ -0,0 +1,4 @@ +DROP FUNCTION CONVERT_STRING_TO_UPPER_CASE; +DROP TABLE USERS; +DROP PROCEDURE CREATE_USER; +DROP PROCEDURE CREATE_USER_RETURN_ALL; \ No newline at end of file