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

Merge pull request #442 from olegz/INT-2544b

Browse Source
This commit is contained in:
Gary Russell
2012-05-21 11:51:12 -04:00
parent a1c2c63534 22536d26bf
commit e1c4281290
24 changed files with 385 additions and 243 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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 <code>org/springframework/integration/jdbc/schema-*.sql</code>, where <code>*</code> 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 <code>DEFAULT</code>.
*
*
* @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<T> 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<Message<?>> messages = new ArrayList<Message<?>>();
final AtomicReference<Date> date = new AtomicReference<Date>();
final AtomicReference<Date> createDate = new AtomicReference<Date>();
final AtomicReference<Date> updateDate = new AtomicReference<Date>();
final AtomicReference<Boolean> completeFlag = new AtomicReference<Boolean>();
final AtomicReference<Integer> lastReleasedSequenceRef = new AtomicReference<Integer>();
final AtomicInteger size = new AtomicInteger();
jdbcTemplate.query(getQuery(LIST_MESSAGES_BY_GROUP_KEY), new Object[] { key, region },
List<Message<?>> 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<Message<?>>() {
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<MessageGroup> iterator() {
final Iterator<String> 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<Message<?>> 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<UUID> getMessageIdsForGroup(Object groupId){
String key = getKey(groupId);
final List<UUID> messageIds = new ArrayList<UUID>();
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<Long> date = new AtomicReference<Long>();
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<Message<?>> {

17
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql
View File

@@ -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
);

17
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql
View File

@@ -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
);

2
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-db2.sql
View File

@@ -2,3 +2,5 @@
DROP TABLE INT_MESSAGE ;
DROP TABLE INT_MESSAGE_GROUP ;
DROP TABLE INT_GROUP_TO_MESSAGE ;
DROP INDEX INT_MESSAGE_IX1 ;

2
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-derby.sql
View File

@@ -2,3 +2,5 @@
DROP TABLE INT_MESSAGE ;
DROP TABLE INT_MESSAGE_GROUP ;
DROP TABLE INT_GROUP_TO_MESSAGE ;
DROP INDEX INT_MESSAGE_IX1 ;

2
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-h2.sql
View File

@@ -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;

2
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-hsqldb.sql
View File

@@ -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;

2
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-mysql.sql
View File

@@ -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 ;

2
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-oracle10g.sql
View File

@@ -2,3 +2,5 @@
DROP TABLE INT_MESSAGE ;
DROP TABLE INT_MESSAGE_GROUP ;
DROP TABLE INT_GROUP_TO_MESSAGE ;
DROP INDEX INT_MESSAGE_IX1 ;

2
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-postgresql.sql
View File

@@ -2,3 +2,5 @@
DROP TABLE INT_MESSAGE ;
DROP TABLE INT_MESSAGE_GROUP ;
DROP TABLE INT_GROUP_TO_MESSAGE ;
DROP INDEX INT_MESSAGE_IX1 ;

2
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sqlserver.sql
View File

@@ -2,3 +2,5 @@
DROP TABLE INT_MESSAGE ;
DROP TABLE INT_MESSAGE_GROUP ;
DROP TABLE INT_GROUP_TO_MESSAGE ;
DROP INDEX INT_MESSAGE_IX1 ;

2
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sybase.sql
View File

@@ -2,3 +2,5 @@
DROP TABLE INT_MESSAGE ;
DROP TABLE INT_MESSAGE_GROUP ;
DROP TABLE INT_GROUP_TO_MESSAGE ;
DROP INDEX INT_MESSAGE_IX1 ;

17
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql
View File

@@ -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
);

17
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql
View File

@@ -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
);

17
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql
View File

@@ -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;

17
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle10g.sql
View File

@@ -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
);

17
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql
View File

@@ -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
);

17
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql
View File

@@ -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
);

19
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql
View File

@@ -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;

2
spring-integration-jdbc/src/main/sql/destroy.sql.vpp
View File

@@ -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};

17
spring-integration-jdbc/src/main/sql/schema.sql.vpp
View File

@@ -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;

51
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java
View File

@@ -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<String> 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<String> 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());
}
}

25
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcOutboundChannelAdapterWithinChainTests-context.xml
View File

@@ -1,17 +1,20 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns: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">
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">
<jdbc:embedded-database id="dataSource" type="DERBY">
<jdbc:script location="classpath:derby-stored-procedures.sql"/>
</jdbc:embedded-database>
<jdbc:embedded-database id="dataSource" type="DERBY"/>
<jdbc:initialize-database data-source="dataSource" ignore-failures="DROPS">
<jdbc:script location="classpath:drop-derby-stored-procedures.sql"/>
<jdbc:script location="classpath:derby-stored-procedures.sql"/>
</jdbc:initialize-database>
<bean id="jdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate">
<constructor-arg ref="dataSource" />

4
spring-integration-jdbc/src/test/resources/drop-derby-stored-procedures.sql Normal file
View File

@@ -0,0 +1,4 @@
DROP FUNCTION CONVERT_STRING_TO_UPPER_CASE;
DROP TABLE USERS;
DROP PROCEDURE CREATE_USER;
DROP PROCEDURE CREATE_USER_RETURN_ALL;