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

INT-1195: rename correlation->group in MessageGroup and friends

Browse Source
This commit is contained in:
David Syer
2010-06-19 06:59:05 +00:00
parent 7d50eb4204
commit 87352799fb
24 changed files with 141 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java
View File

@@ -79,7 +79,7 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag
for (String keyToRemove : conflictKeys) {
if (logger.isDebugEnabled()) {
logger.debug("Excluding header '" + keyToRemove + "' upon aggregation due to conflict(s) "
+ "in MessageGroup with correlation key: " + group.getCorrelationKey());
+ "in MessageGroup with correlation key: " + group.getGroupId());
}
aggregatedHeaders.remove(keyToRemove);
}

4
spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java
View File

@@ -219,7 +219,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
private final boolean forceComplete(MessageGroup group) {
Object correlationKey = group.getCorrelationKey();
Object correlationKey = group.getGroupId();
Object lock = getLock(correlationKey);
synchronized (lock) {
@@ -276,7 +276,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
}
private void remove(MessageGroup group) {
Object correlationKey = group.getCorrelationKey();
Object correlationKey = group.getGroupId();
messageStore.removeMessageGroup(correlationKey);
locks.remove(correlationKey);
}

5
spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroup.java
View File

@@ -30,10 +30,9 @@ public interface MessageGroup {
Collection<Message<?>> getMarked();
/**
* @return the correlation key that links these messages together, typically according
* to a particular CorrelationStrategy
* @return the key that links these messages together
*/
Object getCorrelationKey();
Object getGroupId();
/**
* @return true if the group is complete (i.e. no more messages are expected to be added)

26
spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java
View File

@@ -27,7 +27,7 @@ import org.springframework.integration.core.Message;
/**
* A {@link BlockingQueue} that is backed by a {@link MessageGroupStore}. Can be used to ensure guaranteed delivery in
* the face of transaction rollback (assuming the store is transactional) and also to ensure messages are not lost if
* the process dies (assuming the store is durable). To use the queue across process re-starts, the same correlation key
* the process dies (assuming the store is durable). To use the queue across process re-starts, the same group id
* must be provided, so it needs to be unique but identifiable with a single logical instance of the queue.
*
* @author Dave Syer
@@ -40,7 +40,7 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
private final MessageGroupStore messageGroupStore;
private final Object correlationKey;
private final Object groupId;
private final int capacity;
@@ -53,13 +53,13 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
// This one only needs to be local
private Object readLock = new Object();
public MessageGroupQueue(MessageGroupStore messageGroupStore, Object correlationKey) {
this(messageGroupStore, correlationKey, DEFAULT_CAPACITY);
public MessageGroupQueue(MessageGroupStore messageGroupStore, Object groupId) {
this(messageGroupStore, groupId, DEFAULT_CAPACITY);
}
public MessageGroupQueue(MessageGroupStore messageGroupStore, Object correlationKey, int capacity) {
public MessageGroupQueue(MessageGroupStore messageGroupStore, Object groupId, int capacity) {
this.messageGroupStore = messageGroupStore;
this.correlationKey = correlationKey;
this.groupId = groupId;
this.capacity = capacity;
}
@@ -72,11 +72,11 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
}
public boolean offer(Message<?> e) {
if (messageGroupStore.getMessageGroup(correlationKey).size() >= capacity) {
if (messageGroupStore.getMessageGroup(groupId).size() >= capacity) {
return false;
}
synchronized (storeLock) {
messageGroupStore.addMessageToGroup(correlationKey, e);
messageGroupStore.addMessageToGroup(groupId, e);
}
synchronized (readLock) {
readLock.notifyAll();
@@ -100,7 +100,7 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
return null;
}
result = unmarked.iterator().next();
messageGroupStore.removeMessageFromGroup(correlationKey, result);
messageGroupStore.removeMessageFromGroup(groupId, result);
}
synchronized (writeLock) {
writeLock.notifyAll();
@@ -113,7 +113,7 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
synchronized (storeLock) {
unmarked = getUnmarked();
c.addAll(unmarked);
messageGroupStore.markMessageGroup(messageGroupStore.getMessageGroup(correlationKey));
messageGroupStore.markMessageGroup(messageGroupStore.getMessageGroup(groupId));
}
synchronized (writeLock) {
writeLock.notifyAll();
@@ -127,7 +127,7 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
Iterator<Message<?>> unmarked = getUnmarked().iterator();
for (int i = 0; i < maxElements && unmarked.hasNext(); i++) {
Message<?> message = unmarked.next();
messageGroupStore.removeMessageFromGroup(correlationKey, message);
messageGroupStore.removeMessageFromGroup(groupId, message);
list.add(message);
}
}
@@ -174,7 +174,7 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
}
public int remainingCapacity() {
return capacity - messageGroupStore.getMessageGroup(correlationKey).size();
return capacity - messageGroupStore.getMessageGroup(groupId).size();
}
public Message<?> take() throws InterruptedException {
@@ -189,7 +189,7 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
}
private Collection<Message<?>> getUnmarked() {
return messageGroupStore.getMessageGroup(correlationKey).getUnmarked();
return messageGroupStore.getMessageGroup(groupId).getUnmarked();
}
}

21
spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java
View File

@@ -15,7 +15,7 @@ package org.springframework.integration.store;
import org.springframework.integration.core.Message;
/**
* Interface for storage operations on groups of messages linked by a correlation key.
* Interface for storage operations on groups of messages linked by a group id.
*
* @author Dave Syer
*
@@ -26,20 +26,19 @@ public interface MessageGroupStore {
/**
* Return all Messages currently in the MessageStore that were stored using
* {@link #addMessageToGroup(Object, Message)} with this correlation id.
* {@link #addMessageToGroup(Object, Message)} with this group id.
*
* @return a group of messages, empty if none exists for this key
*/
MessageGroup getMessageGroup(Object correlationKey);
MessageGroup getMessageGroup(Object groupId);
/**
* Store a message with an association to a correlation key. This can be used to group messages together instead of
* storing them just under their id.
* Store a message with an association to a group id. This can be used to group messages together.
*
* @param correlationKey the correlation id to store the message under
* @param groupId the group id to store the message under
* @param message a message
*/
MessageGroup addMessageToGroup(Object correlationKey, Message<?> message);
MessageGroup addMessageToGroup(Object groupId, Message<?> message);
/**
* Persist the mark on all the messages from the group. The group is modified in the process as all its unmarked
@@ -52,17 +51,17 @@ public interface MessageGroupStore {
/**
* Persist a mark on a single message from the group. The group is modified to reflect that 'messageToMark' is no
* longer unmarked but became marked instead.
* @param key the correlationKey for the group containing the message
* @param key the groupId for the group containing the message
* @param messageToMark the message to be marked
*/
MessageGroup removeMessageFromGroup(Object key, Message<?> messageToMark);
/**
* Remove the message group with this correlation key.
* Remove the message group with this id.
*
* @param correlationKey the correlation id to remove
* @param groupId the id of the group to remove
*/
void removeMessageGroup(Object correlationKey);
void removeMessageGroup(Object groupId);
/**
* Register a callback for when a message group is expired through {@link #expireMessageGroups(long)}.

25
spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java
View File

@@ -21,9 +21,8 @@ import java.util.concurrent.LinkedBlockingQueue;
import org.springframework.integration.core.Message;
/**
* Represents a mutable group of correlated messages that is bound to a certain {@link MessageStore} and correlation
* key. The group will grow during its lifetime, when messages are <code>add</code>ed to it. This MessageGroup is thread
* safe.
* Represents a mutable group of correlated messages that is bound to a certain {@link MessageStore} and group id. The
* group will grow during its lifetime, when messages are <code>add</code>ed to it. This MessageGroup is thread safe.
*
* @author Iwein Fuld
* @author Oleg Zhurakousky
@@ -32,7 +31,7 @@ import org.springframework.integration.core.Message;
*/
public class SimpleMessageGroup implements MessageGroup {
private final Object correlationKey;
private final Object groupId;
// Guards(marked, unmarked)
private final Object lock = new Object();
@@ -45,18 +44,18 @@ public class SimpleMessageGroup implements MessageGroup {
private final long timestamp;
public SimpleMessageGroup(Object correlationKey) {
this(Collections.<Message<?>> emptyList(), Collections.<Message<?>> emptyList(), correlationKey, System
public SimpleMessageGroup(Object groupId) {
this(Collections.<Message<?>> emptyList(), Collections.<Message<?>> emptyList(), groupId, System
.currentTimeMillis());
}
public SimpleMessageGroup(Collection<? extends Message<?>> unmarked, Object correlationKey) {
this(unmarked, Collections.<Message<?>> emptyList(), correlationKey, System.currentTimeMillis());
public SimpleMessageGroup(Collection<? extends Message<?>> unmarked, Object groupId) {
this(unmarked, Collections.<Message<?>> emptyList(), groupId, System.currentTimeMillis());
}
public SimpleMessageGroup(Collection<? extends Message<?>> unmarked, Collection<? extends Message<?>> marked,
Object correlationKey, long timestamp) {
this.correlationKey = correlationKey;
Object groupId, long timestamp) {
this.groupId = groupId;
this.timestamp = timestamp;
synchronized (lock) {
for (Message<?> message : unmarked) {
@@ -69,7 +68,7 @@ public class SimpleMessageGroup implements MessageGroup {
}
public SimpleMessageGroup(MessageGroup template) {
this.correlationKey = template.getCorrelationKey();
this.groupId = template.getGroupId();
synchronized (lock) {
// Explicit iteration to work around bug in JDK (before 1.6.0_20
for (Message<?> message : template.getMarked()) {
@@ -135,8 +134,8 @@ public class SimpleMessageGroup implements MessageGroup {
}
}
public Object getCorrelationKey() {
return correlationKey;
public Object getGroupId() {
return groupId;
}
public boolean isComplete() {

34
spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java
View File

@@ -38,7 +38,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes
private final ConcurrentMap<UUID, Message<?>> idToMessage;
private final ConcurrentMap<Object, SimpleMessageGroup> correlationToMessageGroup;
private final ConcurrentMap<Object, SimpleMessageGroup> groupIdToMessageGroup;
private final UpperBound individualUpperBound;
@@ -53,7 +53,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes
*/
public SimpleMessageStore(int individualCapacity, int groupCapacity) {
this.idToMessage = new ConcurrentHashMap<UUID, Message<?>>();
this.correlationToMessageGroup = new ConcurrentHashMap<Object, SimpleMessageGroup>();
this.groupIdToMessageGroup = new ConcurrentHashMap<Object, SimpleMessageGroup>();
this.individualUpperBound = new UpperBound(individualCapacity);
this.groupUpperBound = new UpperBound(groupCapacity);
}
@@ -94,35 +94,35 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes
return null;
}
public MessageGroup getMessageGroup(Object correlationId) {
Assert.notNull(correlationId, "'correlationKey' must not be null");
SimpleMessageGroup group = correlationToMessageGroup.get(correlationId);
public MessageGroup getMessageGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
SimpleMessageGroup group = groupIdToMessageGroup.get(groupId);
if (group == null) {
return new SimpleMessageGroup(correlationId);
return new SimpleMessageGroup(groupId);
}
return new SimpleMessageGroup(group);
}
public MessageGroup addMessageToGroup(Object correlationId, Message<?> message) {
public MessageGroup addMessageToGroup(Object groupId, Message<?> message) {
if (!groupUpperBound.tryAcquire(0)) {
throw new MessagingException(this.getClass().getSimpleName()
+ " was out of capacity at, try constructing it with a larger capacity.");
}
SimpleMessageGroup group = getMessageGroupInternal(correlationId);
SimpleMessageGroup group = getMessageGroupInternal(groupId);
group.add(message);
return group;
}
public MessageGroup markMessageGroup(MessageGroup group) {
Object correlationId = group.getCorrelationKey();
SimpleMessageGroup internal = getMessageGroupInternal(correlationId);
Object groupId = group.getGroupId();
SimpleMessageGroup internal = getMessageGroupInternal(groupId);
internal.markAll();
return internal;
}
public void removeMessageGroup(Object correlationId) {
public void removeMessageGroup(Object groupId) {
groupUpperBound.release();
correlationToMessageGroup.remove(correlationId);
groupIdToMessageGroup.remove(groupId);
}
public MessageGroup removeMessageFromGroup(Object key, Message<?> messageToMark) {
@@ -133,14 +133,14 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes
@Override
public Iterator<MessageGroup> iterator() {
return new HashSet<MessageGroup>(correlationToMessageGroup.values()).iterator();
return new HashSet<MessageGroup>(groupIdToMessageGroup.values()).iterator();
}
private SimpleMessageGroup getMessageGroupInternal(Object correlationId) {
if (!correlationToMessageGroup.containsKey(correlationId)) {
correlationToMessageGroup.putIfAbsent(correlationId, new SimpleMessageGroup(correlationId));
private SimpleMessageGroup getMessageGroupInternal(Object groupId) {
if (!groupIdToMessageGroup.containsKey(groupId)) {
groupIdToMessageGroup.putIfAbsent(groupId, new SimpleMessageGroup(groupId));
}
return correlationToMessageGroup.get(correlationId);
return groupIdToMessageGroup.get(groupId);
}
}

2
spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTest.java
View File

@@ -123,7 +123,7 @@ public class CorrelatingMessageBarrierTest {
public boolean canRelease(MessageGroup messageGroup) {
System.out.println("Trying to release group: " + messageGroup + "\n to thread: " + Thread.currentThread());
Object correlationKey = messageGroup.getCorrelationKey();
Object correlationKey = messageGroup.getGroupId();
Semaphore lock = lockForKey(correlationKey);
System.out.println(Thread.currentThread() + " got lock: " + lock);
return lock.tryAcquire();

2
spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreReaperTests.java
View File

@@ -59,7 +59,7 @@ public class MessageStoreReaperTests {
public void execute(MessageGroupStore messageGroupStore, MessageGroup group) {
groups.add(group);
messageGroupStore.removeMessageGroup(group.getCorrelationKey());
messageGroupStore.removeMessageGroup(group.getGroupId());
}
}

4
spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java
View File

@@ -53,7 +53,7 @@ public class MessageStoreTests {
store.registerMessageGroupExpiryCallback(new MessageGroupCallback() {
public void execute(MessageGroupStore messageGroupStore, MessageGroup group) {
list.add(group.getOne().getPayload().toString());
messageGroupStore.removeMessageGroup(group.getCorrelationKey());
messageGroupStore.removeMessageGroup(group.getGroupId());
}
});
@@ -90,7 +90,7 @@ public class MessageStoreTests {
}
public void removeMessageGroup(Object correlationKey) {
if (correlationKey.equals(testMessages.getCorrelationKey())) {
if (correlationKey.equals(testMessages.getGroupId())) {
removed = true;
}
}

2
spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java
View File

@@ -119,7 +119,7 @@ public class SimpleMessageStoreTests {
store.registerMessageGroupExpiryCallback(new MessageGroupCallback() {
public void execute(MessageGroupStore messageGroupStore, MessageGroup group) {
list.add(group.getOne().getPayload().toString());
messageGroupStore.removeMessageGroup(group.getCorrelationKey());
messageGroupStore.removeMessageGroup(group.getGroupId());
}
});

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

@@ -68,22 +68,22 @@ 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_UNMARKED_MESSAGES_BY_CORRELATION_KEY = "SELECT MESSAGE_ID, CREATED_DATE, CORRELATION_KEY, MESSAGE_BYTES from %PREFIX%MESSAGE_GROUP where CORRELATION_KEY=? and REGION=? and MARKED=0 order by CREATED_DATE";
private static final String LIST_UNMARKED_MESSAGES_BY_GROUP_KEY = "SELECT MESSAGE_ID, CREATED_DATE, GROUP_KEY, MESSAGE_BYTES from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=? and MARKED=0 order by CREATED_DATE";
private static final String LIST_MARKED_MESSAGES_BY_CORRELATION_KEY = "SELECT MESSAGE_ID, CREATED_DATE, CORRELATION_KEY, MESSAGE_BYTES from %PREFIX%MESSAGE_GROUP where CORRELATION_KEY=? and REGION=? and MARKED=1";
private static final String LIST_MARKED_MESSAGES_BY_GROUP_KEY = "SELECT MESSAGE_ID, CREATED_DATE, GROUP_KEY, MESSAGE_BYTES from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=? and MARKED=1";
private static final String GET_MIN_CREATED_DATE_BY_CORRELATION_KEY = "SELECT MIN(CREATED_DATE) from %PREFIX%MESSAGE_GROUP where CORRELATION_KEY=? and REGION=?";
private static final String GET_MIN_CREATED_DATE_BY_GROUP_KEY = "SELECT MIN(CREATED_DATE) from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=?";
private static final String MARK_MESSAGES_IN_GROUP = "UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=?, MARKED=1 where MARKED=0 and CORRELATION_KEY=? and REGION=?";
private static final String MARK_MESSAGES_IN_GROUP = "UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=?, MARKED=1 where MARKED=0 and GROUP_KEY=? and REGION=?";
private static final String REMOVE_MESSAGE_FROM_GROUP = "DELETE from %PREFIX%MESSAGE_GROUP where CORRELATION_KEY=? and REGION=? and MESSAGE_ID=?";
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 DELETE_MESSAGE_GROUP = "DELETE from %PREFIX%MESSAGE_GROUP where CORRELATION_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, CORRELATION_KEY, MARKED, MESSAGE_BYTES)"
private static final String CREATE_MESSAGE_IN_GROUP = "INSERT into %PREFIX%MESSAGE_GROUP(MESSAGE_ID, REGION, CREATED_DATE, GROUP_KEY, MARKED, MESSAGE_BYTES)"
+ " values (?, ?, ?, ?, 0, ?)";
private static final String LIST_CORRELATION_KEYS = "SELECT distinct CORRELATION_KEY as CREATED from %PREFIX%MESSAGE_GROUP where REGION=?";
private static final String LIST_GROUP_KEYS = "SELECT distinct GROUP_KEY as CREATED from %PREFIX%MESSAGE_GROUP where REGION=?";
public static final int DEFAULT_LONG_STRING_LENGTH = 2500;
@@ -242,11 +242,11 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
return result;
}
public MessageGroup addMessageToGroup(Object correlationKey, Message<?> message) {
public MessageGroup addMessageToGroup(Object groupId, Message<?> message) {
final long createdDate = System.currentTimeMillis();
final String messageId = getKey(message.getHeaders().getId());
final String correlationId = getKey(correlationKey);
final String groupKey = getKey(groupId);
final byte[] messageBytes = SerializationUtils.serialize(message);
jdbcTemplate.update(getQuery(CREATE_MESSAGE_IN_GROUP), new PreparedStatementSetter() {
@@ -255,72 +255,72 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
ps.setString(1, messageId);
ps.setString(2, region);
ps.setTimestamp(3, new Timestamp(createdDate));
ps.setString(4, correlationId);
ps.setString(4, groupKey);
lobHandler.getLobCreator().setBlobAsBytes(ps, 5, messageBytes);
}
});
return getMessageGroup(correlationKey);
return getMessageGroup(groupId);
}
public MessageGroup getMessageGroup(Object correlationKey) {
String key = getKey(correlationKey);
List<Message<?>> marked = jdbcTemplate.query(getQuery(LIST_MARKED_MESSAGES_BY_CORRELATION_KEY), new Object[] {
public MessageGroup getMessageGroup(Object groupId) {
String key = getKey(groupId);
List<Message<?>> marked = jdbcTemplate.query(getQuery(LIST_MARKED_MESSAGES_BY_GROUP_KEY), new Object[] {
key, region }, mapper);
List<Message<?>> unmarked = jdbcTemplate.query(getQuery(LIST_UNMARKED_MESSAGES_BY_CORRELATION_KEY),
List<Message<?>> unmarked = jdbcTemplate.query(getQuery(LIST_UNMARKED_MESSAGES_BY_GROUP_KEY),
new Object[] { key, region }, mapper);
if (marked.isEmpty() && unmarked.isEmpty()) {
return new SimpleMessageGroup(correlationKey);
return new SimpleMessageGroup(groupId);
}
Timestamp date = jdbcTemplate.queryForObject(getQuery(GET_MIN_CREATED_DATE_BY_CORRELATION_KEY),
Timestamp date = jdbcTemplate.queryForObject(getQuery(GET_MIN_CREATED_DATE_BY_GROUP_KEY),
Timestamp.class, key, region);
Assert.state(date != null, "Could not locate created date for correlationKey=" + correlationKey);
Assert.state(date != null, "Could not locate created date for groupId=" + groupId);
long timestamp = date.getTime();
return new SimpleMessageGroup(unmarked, marked, correlationKey, timestamp);
return new SimpleMessageGroup(unmarked, marked, groupId, timestamp);
}
public MessageGroup markMessageGroup(MessageGroup group) {
final long updatedDate = System.currentTimeMillis();
final String correlationId = getKey(group.getCorrelationKey());
final String groupKey = getKey(group.getGroupId());
jdbcTemplate.update(getQuery(MARK_MESSAGES_IN_GROUP), new PreparedStatementSetter() {
public void setValues(PreparedStatement ps) throws SQLException {
logger.debug("Marking messages with correlation key=" + correlationId);
logger.debug("Marking messages with group key=" + groupKey);
ps.setTimestamp(1, new Timestamp(updatedDate));
ps.setString(2, correlationId);
ps.setString(2, groupKey);
ps.setString(3, region);
}
});
return getMessageGroup(group.getCorrelationKey());
return getMessageGroup(group.getGroupId());
}
public MessageGroup removeMessageFromGroup(Object correlationKey, Message<?> messageToMark) {
final String correlationId = getKey(correlationKey);
public MessageGroup removeMessageFromGroup(Object groupId, Message<?> messageToMark) {
final String groupKey = getKey(groupId);
final String messageId = getKey(messageToMark.getHeaders().getId());
jdbcTemplate.update(getQuery(REMOVE_MESSAGE_FROM_GROUP), new PreparedStatementSetter() {
public void setValues(PreparedStatement ps) throws SQLException {
logger.debug("Removing message from group with correlation key=" + correlationId);
ps.setString(1, correlationId);
logger.debug("Removing message from group with group key=" + groupKey);
ps.setString(1, groupKey);
ps.setString(2, region);
ps.setString(3, messageId);
}
});
return getMessageGroup(correlationKey);
return getMessageGroup(groupId);
}
public void removeMessageGroup(Object correlationKey) {
public void removeMessageGroup(Object groupId) {
final String correlationId = getKey(correlationKey);
final String groupKey = getKey(groupId);
jdbcTemplate.update(getQuery(DELETE_MESSAGE_GROUP), new PreparedStatementSetter() {
public void setValues(PreparedStatement ps) throws SQLException {
logger.debug("Marking messages with correlation key=" + correlationId);
ps.setString(1, correlationId);
logger.debug("Marking messages with group key=" + groupKey);
ps.setString(1, groupKey);
ps.setString(2, region);
}
});
@@ -331,7 +331,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
public Iterator<MessageGroup> iterator() {
@SuppressWarnings("unchecked")
final Iterator<String> iterator = jdbcTemplate.query(getQuery(LIST_CORRELATION_KEYS), new Object[] { region },
final Iterator<String> iterator = jdbcTemplate.query(getQuery(LIST_GROUP_KEYS), new Object[] { region },
new SingleColumnRowMapper(String.class)).iterator();
return new Iterator<MessageGroup>() {

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

@@ -9,11 +9,11 @@ CREATE TABLE INT_MESSAGE (
CREATE TABLE INT_MESSAGE_GROUP (
MESSAGE_ID VARCHAR(100) NOT NULL,
CORRELATION_KEY VARCHAR(100) NOT NULL,
GROUP_KEY VARCHAR(100) NOT NULL,
REGION VARCHAR(100),
MARKED BIGINT,
CREATED_DATE TIMESTAMP NOT NULL,
UPDATED_DATE TIMESTAMP,
MESSAGE_BYTES BLOB,
constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, CORRELATION_KEY)
constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, GROUP_KEY)
);

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

@@ -9,11 +9,11 @@ CREATE TABLE INT_MESSAGE (
CREATE TABLE INT_MESSAGE_GROUP (
MESSAGE_ID VARCHAR(100) NOT NULL,
CORRELATION_KEY VARCHAR(100) NOT NULL,
GROUP_KEY VARCHAR(100) NOT NULL,
REGION VARCHAR(100),
MARKED BIGINT,
CREATED_DATE TIMESTAMP NOT NULL,
UPDATED_DATE TIMESTAMP,
MESSAGE_BYTES BLOB,
constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, CORRELATION_KEY)
constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, GROUP_KEY)
);

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

@@ -9,11 +9,11 @@ CREATE TABLE INT_MESSAGE (
CREATE TABLE INT_MESSAGE_GROUP (
MESSAGE_ID VARCHAR(100) NOT NULL,
CORRELATION_KEY VARCHAR(100) NOT NULL,
GROUP_KEY VARCHAR(100) NOT NULL,
REGION VARCHAR(100),
MARKED BIGINT,
CREATED_DATE TIMESTAMP NOT NULL,
UPDATED_DATE TIMESTAMP,
MESSAGE_BYTES LONGVARBINARY,
constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, CORRELATION_KEY)
constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, GROUP_KEY)
);

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

@@ -9,11 +9,11 @@ CREATE TABLE INT_MESSAGE (
CREATE TABLE INT_MESSAGE_GROUP (
MESSAGE_ID VARCHAR(100) NOT NULL,
CORRELATION_KEY VARCHAR(100) NOT NULL,
GROUP_KEY VARCHAR(100) NOT NULL,
REGION VARCHAR(100),
MARKED BIGINT,
CREATED_DATE TIMESTAMP NOT NULL,
UPDATED_DATE TIMESTAMP,
MESSAGE_BYTES LONGVARBINARY,
constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, CORRELATION_KEY)
constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, GROUP_KEY)
);

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

@@ -9,11 +9,11 @@ CREATE TABLE INT_MESSAGE (
CREATE TABLE INT_MESSAGE_GROUP (
MESSAGE_ID VARCHAR(100) NOT NULL,
CORRELATION_KEY VARCHAR(100) NOT NULL,
GROUP_KEY VARCHAR(100) NOT NULL,
REGION VARCHAR(100),
MARKED BIGINT,
CREATED_DATE DATETIME NOT NULL,
UPDATED_DATE DATETIME,
MESSAGE_BYTES BLOB,
constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, CORRELATION_KEY)
constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, GROUP_KEY)
);

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

@@ -9,11 +9,11 @@ CREATE TABLE INT_MESSAGE (
CREATE TABLE INT_MESSAGE_GROUP (
MESSAGE_ID VARCHAR2(100) NOT NULL,
CORRELATION_KEY VARCHAR2(100) NOT NULL,
GROUP_KEY VARCHAR2(100) NOT NULL,
REGION VARCHAR2(100),
MARKED NUMBER(19,0),
CREATED_DATE TIMESTAMP NOT NULL,
UPDATED_DATE TIMESTAMP,
MESSAGE_BYTES BLOB,
constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, CORRELATION_KEY)
constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, GROUP_KEY)
);

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

@@ -9,11 +9,11 @@ CREATE TABLE INT_MESSAGE (
CREATE TABLE INT_MESSAGE_GROUP (
MESSAGE_ID VARCHAR(100) NOT NULL,
CORRELATION_KEY VARCHAR(100) NOT NULL,
GROUP_KEY VARCHAR(100) NOT NULL,
REGION VARCHAR(100),
MARKED BIGINT,
CREATED_DATE TIMESTAMP NOT NULL,
UPDATED_DATE TIMESTAMP,
MESSAGE_BYTES BYTEA,
constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, CORRELATION_KEY)
constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, GROUP_KEY)
);

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

@@ -9,11 +9,11 @@ CREATE TABLE INT_MESSAGE (
CREATE TABLE INT_MESSAGE_GROUP (
MESSAGE_ID VARCHAR(100) NOT NULL,
CORRELATION_KEY VARCHAR(100) NOT NULL,
GROUP_KEY VARCHAR(100) NOT NULL,
REGION VARCHAR(100),
MARKED BIGINT,
CREATED_DATE DATETIME NOT NULL,
UPDATED_DATE DATETIME,
MESSAGE_BYTES IMAGE,
constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, CORRELATION_KEY)
constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, GROUP_KEY)
);

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

@@ -9,11 +9,11 @@ CREATE TABLE INT_MESSAGE (
CREATE TABLE INT_MESSAGE_GROUP (
MESSAGE_ID VARCHAR(100) NOT NULL,
CORRELATION_KEY VARCHAR(100) NOT NULL,
GROUP_KEY VARCHAR(100) NOT NULL,
REGION VARCHAR(100),
MARKED BIGINT,
CREATED_DATE DATETIME NOT NULL,
UPDATED_DATE DATETIME,
MESSAGE_BYTES IMAGE,
constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, CORRELATION_KEY)
constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, GROUP_KEY)
);

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

@@ -9,11 +9,11 @@ CREATE TABLE INT_MESSAGE (
CREATE TABLE INT_MESSAGE_GROUP (
MESSAGE_ID ${VARCHAR}(100) NOT NULL,
CORRELATION_KEY ${VARCHAR}(100) NOT NULL,
GROUP_KEY ${VARCHAR}(100) NOT NULL,
REGION ${VARCHAR}(100),
MARKED ${BIGINT},
CREATED_DATE ${TIMESTAMP} NOT NULL,
UPDATED_DATE ${TIMESTAMP},
MESSAGE_BYTES ${BLOB},
constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, CORRELATION_KEY)
constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, GROUP_KEY)
);

2
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelTests.java
View File

@@ -31,7 +31,7 @@ public class JdbcMessageStoreChannelTests {
@Before
public void clear() {
for (MessageGroup group : messageStore) {
messageStore.removeMessageGroup(group.getCorrelationKey());
messageStore.removeMessageGroup(group.getGroupId());
}
}

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

@@ -112,11 +112,11 @@ public class JdbcMessageStoreTests {
@Test
@Transactional
public void testAddAndGetMessageGroup() throws Exception {
String correlationId = "X";
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(correlationId).build();
String groupId = "X";
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build();
long now = System.currentTimeMillis();
messageStore.addMessageToGroup(correlationId, message);
MessageGroup group = messageStore.getMessageGroup(correlationId);
messageStore.addMessageToGroup(groupId, message);
MessageGroup group = messageStore.getMessageGroup(groupId);
assertEquals(1, group.size());
assertTrue("Timestamp too early: " + group.getTimestamp() + "<" + now, group.getTimestamp() >= now);
}
@@ -124,23 +124,23 @@ public class JdbcMessageStoreTests {
@Test
@Transactional
public void testAddAndRemoveMessageFromMessageGroup() throws Exception {
String correlationId = "X";
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(correlationId).build();
messageStore.addMessageToGroup(correlationId, message);
messageStore.removeMessageFromGroup(correlationId, message);
MessageGroup group = messageStore.getMessageGroup(correlationId);
String groupId = "X";
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build();
messageStore.addMessageToGroup(groupId, message);
messageStore.removeMessageFromGroup(groupId, message);
MessageGroup group = messageStore.getMessageGroup(groupId);
assertEquals(0, group.size());
}
@Test
@Transactional
public void testOrderInMessageGroup() throws Exception {
String correlationId = "X";
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(correlationId).build();
messageStore.addMessageToGroup(correlationId, message);
message = MessageBuilder.withPayload("bar").setCorrelationId(correlationId).build();
messageStore.addMessageToGroup(correlationId, message);
MessageGroup group = messageStore.getMessageGroup(correlationId);
String groupId = "X";
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build();
messageStore.addMessageToGroup(groupId, message);
message = MessageBuilder.withPayload("bar").setCorrelationId(groupId).build();
messageStore.addMessageToGroup(groupId, message);
MessageGroup group = messageStore.getMessageGroup(groupId);
assertEquals(2, group.size());
Iterator<Message<?>> iterator = group.getUnmarked().iterator();
assertEquals("foo", iterator.next().getPayload());
@@ -150,10 +150,10 @@ public class JdbcMessageStoreTests {
@Test
@Transactional
public void testAddAndMarkMessageGroup() throws Exception {
String correlationId = "X";
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(correlationId).build();
messageStore.addMessageToGroup(correlationId, message);
MessageGroup group = messageStore.getMessageGroup(correlationId);
String groupId = "X";
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build();
messageStore.addMessageToGroup(groupId, message);
MessageGroup group = messageStore.getMessageGroup(groupId);
group = messageStore.markMessageGroup(group);
assertEquals(1, group.getMarked().size());
}
@@ -161,16 +161,16 @@ public class JdbcMessageStoreTests {
@Test
@Transactional
public void testExpireMessageGroup() throws Exception {
String correlationId = "X";
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(correlationId).build();
messageStore.addMessageToGroup(correlationId, message);
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.getCorrelationKey());
messageGroupStore.removeMessageGroup(group.getGroupId());
}
});
messageStore.expireMessageGroups(-10000);
MessageGroup group = messageStore.getMessageGroup(correlationId);
MessageGroup group = messageStore.getMessageGroup(groupId);
assertEquals(0, group.size());
}