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 3e09169b86..020ebe4d98 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 @@ -20,6 +20,7 @@ import java.sql.Timestamp; import java.sql.Types; import java.util.ArrayList; import java.util.Date; +import java.util.HashMap; import java.util.Iterator; import java.util.List; import java.util.Map; @@ -30,7 +31,6 @@ 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; @@ -65,6 +65,7 @@ import org.springframework.util.StringUtils; * * @author Dave Syer * @author Oleg Zhurakousky + * @author Matt Stine * @since 2.0 */ @ManagedResource @@ -77,65 +78,77 @@ 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 enum Query { + 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" + + 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=?"; + + " values (?, ?, 0, 0, 0, ?, ?)"), - private static final String REMOVE_MESSAGE_FROM_GROUP = "DELETE from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY=? and MESSAGE_ID=?"; + UPDATE_MESSAGE_GROUP("UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=? where GROUP_KEY=? and REGION=?"), - private static final String COUNT_ALL_MESSAGES_IN_GROUPS = "SELECT COUNT(MESSAGE_ID) from %PREFIX%GROUP_TO_MESSAGE"; + REMOVE_MESSAGE_FROM_GROUP("DELETE from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY=? and MESSAGE_ID=?"), - private static final String COUNT_ALL_MESSAGES_IN_GROUP = "SELECT COUNT(MESSAGE_ID) from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY=?"; + COUNT_ALL_MESSAGES_IN_GROUPS("SELECT COUNT(MESSAGE_ID) from %PREFIX%GROUP_TO_MESSAGE"), - 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"; + COUNT_ALL_MESSAGES_IN_GROUP("SELECT COUNT(MESSAGE_ID) from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY=?"), - 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"; + 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 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 = ?))"; + 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 GET_GROUP_INFO = "SELECT COMPLETE, LAST_RELEASED_SEQUENCE, CREATED_DATE, UPDATED_DATE" + - " from %PREFIX%MESSAGE_GROUP where GROUP_KEY = ?"; + 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_MESSAGE = "SELECT MESSAGE_ID, CREATED_DATE, MESSAGE_BYTES from %PREFIX%MESSAGE where MESSAGE_ID=? and REGION=?"; + GET_GROUP_INFO("SELECT COMPLETE, LAST_RELEASED_SEQUENCE, CREATED_DATE, UPDATED_DATE" + + " from %PREFIX%MESSAGE_GROUP where GROUP_KEY = ?"), - private static final String GET_GROUP_CREATED_DATE = "SELECT CREATED_DATE from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=?"; + GET_MESSAGE("SELECT MESSAGE_ID, CREATED_DATE, MESSAGE_BYTES from %PREFIX%MESSAGE where MESSAGE_ID=? and REGION=?"), - private static final String GET_MESSAGE_COUNT = "SELECT COUNT(MESSAGE_ID) from %PREFIX%MESSAGE where REGION=?"; + GET_GROUP_CREATED_DATE("SELECT CREATED_DATE from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=?"), - private static final String DELETE_MESSAGE = "DELETE from %PREFIX%MESSAGE where MESSAGE_ID=? and REGION=?"; + GET_MESSAGE_COUNT("SELECT COUNT(MESSAGE_ID) from %PREFIX%MESSAGE where REGION=?"), - private static final String CREATE_MESSAGE = "INSERT into %PREFIX%MESSAGE(MESSAGE_ID, REGION, CREATED_DATE, MESSAGE_BYTES)" - + " values (?, ?, ?, ?)"; + DELETE_MESSAGE("DELETE from %PREFIX%MESSAGE where MESSAGE_ID=? and REGION=?"), - private static final String COUNT_ALL_GROUPS = "SELECT COUNT(GROUP_KEY) from %PREFIX%MESSAGE_GROUP where REGION=?"; + CREATE_MESSAGE("INSERT into %PREFIX%MESSAGE(MESSAGE_ID, REGION, CREATED_DATE, MESSAGE_BYTES)" + + " values (?, ?, ?, ?)"), - private static final String COMPLETE_GROUP = "UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=?, COMPLETE=1 where GROUP_KEY=? and REGION=?"; + COUNT_ALL_GROUPS("SELECT COUNT(GROUP_KEY) from %PREFIX%MESSAGE_GROUP where REGION=?"), - private static final String UPDATE_LAST_RELEASED_SEQUENCE = "UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=?, LAST_RELEASED_SEQUENCE=? where GROUP_KEY=? and REGION=?"; + COMPLETE_GROUP("UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=?, COMPLETE=1 where GROUP_KEY=? and REGION=?"), - private static final String DELETE_MESSAGE_GROUP = "DELETE from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=?"; + UPDATE_LAST_RELEASED_SEQUENCE("UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=?, LAST_RELEASED_SEQUENCE=? where GROUP_KEY=? and REGION=?"), - private static final String CREATE_GROUP_TO_MESSAGE = "INSERT into %PREFIX%GROUP_TO_MESSAGE" + - "(GROUP_KEY, MESSAGE_ID)" - + " values (?, ?)"; + DELETE_MESSAGE_GROUP("DELETE from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=?"), - private static final String UPDATE_GROUP = "UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=? where GROUP_KEY=? and REGION=?"; + CREATE_GROUP_TO_MESSAGE("INSERT into %PREFIX%GROUP_TO_MESSAGE" + + "(GROUP_KEY, MESSAGE_ID)" + + " values (?, ?)"), - private static final String LIST_GROUP_KEYS = "SELECT distinct GROUP_KEY as CREATED from %PREFIX%MESSAGE_GROUP where REGION=?"; + UPDATE_GROUP("UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=? where GROUP_KEY=? and REGION=?"), + + LIST_GROUP_KEYS("SELECT distinct GROUP_KEY as CREATED from %PREFIX%MESSAGE_GROUP where REGION=?"); + + private String sql; + + Query(String sql) { + this.sql = sql; + } + + public String getSql() { + return sql; + } + } public static final int DEFAULT_LONG_STRING_LENGTH = 2500; @@ -164,6 +177,8 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa private volatile MessageMapper mapper = new MessageMapper(); + private volatile Map queryCache = new HashMap(); + /** * Convenient constructor for configuration use. */ @@ -266,7 +281,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa if (message == null) { return null; } - int updated = jdbcTemplate.update(getQuery(DELETE_MESSAGE), new Object[] { getKey(id), region }, new int[] { + int updated = jdbcTemplate.update(getQuery(Query.DELETE_MESSAGE), new Object[] { getKey(id), region }, new int[] { Types.VARCHAR, Types.VARCHAR }); if (updated != 0) { return message; @@ -276,11 +291,11 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa @ManagedAttribute public long getMessageCount() { - return jdbcTemplate.queryForInt(getQuery(GET_MESSAGE_COUNT), region); + return jdbcTemplate.queryForInt(getQuery(Query.GET_MESSAGE_COUNT), region); } public Message getMessage(UUID id) { - List> list = jdbcTemplate.query(getQuery(GET_MESSAGE), new Object[] { getKey(id), region }, mapper); + List> list = jdbcTemplate.query(getQuery(Query.GET_MESSAGE), new Object[] { getKey(id), region }, mapper); if (list.isEmpty()) { return null; } @@ -309,7 +324,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa final String messageId = getKey(result.getHeaders().getId()); final byte[] messageBytes = serializer.convert(result); - jdbcTemplate.update(getQuery(CREATE_MESSAGE), new PreparedStatementSetter() { + jdbcTemplate.update(getQuery(Query.CREATE_MESSAGE), new PreparedStatementSetter() { public void setValues(PreparedStatement ps) throws SQLException { if (logger.isDebugEnabled()){ logger.debug("Inserting message with id key=" + messageId); @@ -326,13 +341,13 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa public MessageGroup addMessageToGroup(Object groupId, Message message) { final String groupKey = getKey(groupId); final String messageId = getKey(message.getHeaders().getId()); - boolean groupNotExist = jdbcTemplate.queryForInt(this.getQuery(GROUP_EXISTS), groupKey) < 1; + boolean groupNotExist = jdbcTemplate.queryForInt(this.getQuery(Query.GROUP_EXISTS), groupKey) < 1; 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); + jdbcTemplate.queryForObject(getQuery(Query.GET_GROUP_CREATED_DATE), new Object[] { groupKey, region}, Timestamp.class); if (groupNotExist){ try { @@ -348,7 +363,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa this.addMessage(message); - jdbcTemplate.update(getQuery(CREATE_GROUP_TO_MESSAGE), new PreparedStatementSetter() { + jdbcTemplate.update(getQuery(Query.CREATE_GROUP_TO_MESSAGE), new PreparedStatementSetter() { public void setValues(PreparedStatement ps) throws SQLException { if (logger.isDebugEnabled()){ logger.debug("Inserting message with id key=" + messageId + " and created date=" + createdDate); @@ -364,19 +379,19 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa @Override @ManagedAttribute public int getMessageGroupCount() { - return jdbcTemplate.queryForInt(getQuery(COUNT_ALL_GROUPS), region); + return jdbcTemplate.queryForInt(getQuery(Query.COUNT_ALL_GROUPS), region); } @Override @ManagedAttribute public int getMessageCountForAllMessageGroups() { - return jdbcTemplate.queryForInt(getQuery(COUNT_ALL_MESSAGES_IN_GROUPS)); + return jdbcTemplate.queryForInt(getQuery(Query.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); + return jdbcTemplate.queryForInt(getQuery(Query.COUNT_ALL_MESSAGES_IN_GROUP), key); } public MessageGroup getMessageGroup(Object groupId) { @@ -386,9 +401,9 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa final AtomicReference completeFlag = new AtomicReference(); final AtomicReference lastReleasedSequenceRef = new AtomicReference(); - List> messages = jdbcTemplate.query(getQuery(LIST_MESSAGES_BY_GROUP_KEY), new Object[] { key, region }, mapper); + List> messages = jdbcTemplate.query(getQuery(Query.LIST_MESSAGES_BY_GROUP_KEY), new Object[] { key, region }, mapper); - jdbcTemplate.query(getQuery(GET_GROUP_INFO), new Object[] { key}, + jdbcTemplate.query(getQuery(Query.GET_GROUP_INFO), new Object[] { key}, new RowCallbackHandler() { public void processRow(ResultSet rs) throws SQLException { updateDate.set(rs.getTimestamp("UPDATED_DATE")); @@ -423,7 +438,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa final String groupKey = getKey(groupId); final String messageId = getKey(messageToRemove.getHeaders().getId()); - jdbcTemplate.update(getQuery(REMOVE_MESSAGE_FROM_GROUP), new PreparedStatementSetter() { + jdbcTemplate.update(getQuery(Query.REMOVE_MESSAGE_FROM_GROUP), new PreparedStatementSetter() { public void setValues(PreparedStatement ps) throws SQLException { if (logger.isDebugEnabled()){ logger.debug("Removing message from group with group key=" + groupKey); @@ -445,7 +460,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa this.removeMessage(messageIds); } - jdbcTemplate.update(getQuery(DELETE_MESSAGE_GROUP), new PreparedStatementSetter() { + jdbcTemplate.update(getQuery(Query.DELETE_MESSAGE_GROUP), new PreparedStatementSetter() { public void setValues(PreparedStatement ps) throws SQLException { if (logger.isDebugEnabled()){ logger.debug("Marking messages with group key=" + groupKey); @@ -460,7 +475,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa final long updatedDate = System.currentTimeMillis(); final String groupKey = getKey(groupId); - jdbcTemplate.update(getQuery(COMPLETE_GROUP), new PreparedStatementSetter() { + jdbcTemplate.update(getQuery(Query.COMPLETE_GROUP), new PreparedStatementSetter() { public void setValues(PreparedStatement ps) throws SQLException { if (logger.isDebugEnabled()){ logger.debug("Completing MessageGroup: " + groupKey); @@ -477,7 +492,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa final long updatedDate = System.currentTimeMillis(); final String groupKey = getKey(groupId); - jdbcTemplate.update(getQuery(UPDATE_LAST_RELEASED_SEQUENCE), new PreparedStatementSetter() { + jdbcTemplate.update(getQuery(Query.UPDATE_LAST_RELEASED_SEQUENCE), new PreparedStatementSetter() { public void setValues(PreparedStatement ps) throws SQLException { if (logger.isDebugEnabled()){ logger.debug("Updating the sequence number of the last released Message in the MessageGroup: " + groupKey); @@ -503,7 +518,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa public Iterator iterator() { - final Iterator iterator = jdbcTemplate.query(getQuery(LIST_GROUP_KEYS), new Object[] { region }, + final Iterator iterator = jdbcTemplate.query(getQuery(Query.LIST_GROUP_KEYS), new Object[] { region }, new SingleColumnRowMapper()).iterator(); return new Iterator() { @@ -523,20 +538,29 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa } /** - * Replace patterns in the input to produce a valid SQL query. This implementation replaces the table prefix. + * 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 base the SQL query to be transformed * @return a transformed query with replacements */ - protected String getQuery(String base) { - return StringUtils.replace(base, "%PREFIX%", tablePrefix); + protected String getQuery(Query base) { + String query = queryCache.get(base); + + if (query == null) { + query = StringUtils.replace(base.getSql(), "%PREFIX%", tablePrefix); + queryCache.put(base, query); + } + + return query; } /** * To be used to get a reference to JdbcOperations * in case this class is subclassed * - * @return + * @return the JdbcOperations implementation */ protected JdbcOperations getJdbcOperations(){ return this.jdbcTemplate; @@ -546,11 +570,11 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa * 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 + * @param groupIdKey String representation of message group ID + * @return a message; 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); + List> messages = jdbcTemplate.query(getQuery(Query.POLL_FROM_GROUP), new Object[] { groupIdKey, region }, mapper); Assert.isTrue(messages.size() == 0 || messages.size() == 1); if (messages.size() > 0){ return messages.get(0); @@ -559,7 +583,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa } private void doCreateMessageGroup(final String groupKey, final Timestamp createdDate){ - jdbcTemplate.update(getQuery(CREATE_MESSAGE_GROUP), new PreparedStatementSetter() { + jdbcTemplate.update(getQuery(Query.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); @@ -573,7 +597,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa } private void doUpdateMessageGroup(final String groupKey, final Timestamp updatedDate){ - jdbcTemplate.update(getQuery(UPDATE_MESSAGE_GROUP), new PreparedStatementSetter() { + jdbcTemplate.update(getQuery(Query.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); @@ -586,7 +610,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa } private void updateMessageGroup(final String groupId){ - jdbcTemplate.update(getQuery(UPDATE_GROUP), new PreparedStatementSetter() { + jdbcTemplate.update(getQuery(Query.UPDATE_GROUP), new PreparedStatementSetter() { public void setValues(PreparedStatement ps) throws SQLException { if (logger.isDebugEnabled()){ logger.debug("Updating MessageGroup: " + groupId); @@ -603,7 +627,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa final List messageIds = new ArrayList(); - jdbcTemplate.query(getQuery(LIST_MESSAGEIDS_BY_GROUP_KEY), new Object[] { key, region }, + jdbcTemplate.query(getQuery(Query.LIST_MESSAGEIDS_BY_GROUP_KEY), new Object[] { key, region }, new RowCallbackHandler() { public void processRow(ResultSet rs) throws SQLException {