Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #500 from mstine/INT-2575

Browse Source
This commit is contained in:
Gary Russell
2012-06-18 17:32:31 -04:00
parent 72562c7aa4 77b8beabaa
commit 41d73d91b3
1 changed files with 91 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

158
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java
View File

@@ -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<Query, String> queryCache = new HashMap<Query, String>();
/**
* 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<Message<?>> list = jdbcTemplate.query(getQuery(GET_MESSAGE), new Object[] { getKey(id), region }, mapper);
List<Message<?>> 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<Boolean> completeFlag = new AtomicReference<Boolean>();
final AtomicReference<Integer> lastReleasedSequenceRef = new AtomicReference<Integer>();
List<Message<?>> messages = jdbcTemplate.query(getQuery(LIST_MESSAGES_BY_GROUP_KEY), new Object[] { key, region }, mapper);
List<Message<?>> 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<MessageGroup> iterator() {
final Iterator<String> iterator = jdbcTemplate.query(getQuery(LIST_GROUP_KEYS), new Object[] { region },
final Iterator<String> iterator = jdbcTemplate.query(getQuery(Query.LIST_GROUP_KEYS), new Object[] { region },
new SingleColumnRowMapper<String>()).iterator();
return new Iterator<MessageGroup>() {
@@ -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<Message<?>> messages = jdbcTemplate.query(getQuery(POLL_FROM_GROUP), new Object[] { groupIdKey, region }, mapper);
List<Message<?>> 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<UUID> messageIds = new ArrayList<UUID>();
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 {