From 87352799fb64385d756f339f5e929dac512da1db Mon Sep 17 00:00:00 2001 From: David Syer Date: Sat, 19 Jun 2010 06:59:05 +0000 Subject: [PATCH] INT-1195: rename correlation->group in MessageGroup and friends --- ...tractAggregatingMessageGroupProcessor.java | 2 +- .../aggregator/CorrelatingMessageHandler.java | 4 +- .../integration/store/MessageGroup.java | 5 +- .../integration/store/MessageGroupQueue.java | 26 +++---- .../integration/store/MessageGroupStore.java | 21 +++--- .../integration/store/SimpleMessageGroup.java | 25 ++++--- .../integration/store/SimpleMessageStore.java | 34 +++++----- .../CorrelatingMessageBarrierTest.java | 2 +- .../store/MessageStoreReaperTests.java | 2 +- .../integration/store/MessageStoreTests.java | 4 +- .../store/SimpleMessageStoreTests.java | 2 +- .../integration/jdbc/JdbcMessageStore.java | 68 +++++++++---------- .../integration/jdbc/schema-db2.sql | 4 +- .../integration/jdbc/schema-derby.sql | 4 +- .../integration/jdbc/schema-h2.sql | 4 +- .../integration/jdbc/schema-hsqldb.sql | 4 +- .../integration/jdbc/schema-mysql.sql | 4 +- .../integration/jdbc/schema-oracle10g.sql | 4 +- .../integration/jdbc/schema-postgresql.sql | 4 +- .../integration/jdbc/schema-sqlserver.sql | 4 +- .../integration/jdbc/schema-sybase.sql | 4 +- .../src/main/sql/schema.sql.vpp | 4 +- .../jdbc/JdbcMessageStoreChannelTests.java | 2 +- .../jdbc/JdbcMessageStoreTests.java | 48 ++++++------- 24 files changed, 141 insertions(+), 144 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java index f51d3599a3..e3107b344a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java @@ -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); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java index fb23937b8c..7c22973f93 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java @@ -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); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroup.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroup.java index 47deceac8a..0fcdc8c05b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroup.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroup.java @@ -30,10 +30,9 @@ public interface MessageGroup { Collection> 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) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java index 365f5b14af..b7150a64e0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java @@ -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> 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> 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> 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> 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> 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> implements Bloc Iterator> 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> 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> implements Bloc } private Collection> getUnmarked() { - return messageGroupStore.getMessageGroup(correlationKey).getUnmarked(); + return messageGroupStore.getMessageGroup(groupId).getUnmarked(); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java index 4c03e17b3e..c2998e2619 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java @@ -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)}. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java index d29f4bed09..ca3d0932ef 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java @@ -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 added 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 added 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.> emptyList(), Collections.> emptyList(), correlationKey, System + public SimpleMessageGroup(Object groupId) { + this(Collections.> emptyList(), Collections.> emptyList(), groupId, System .currentTimeMillis()); } - public SimpleMessageGroup(Collection> unmarked, Object correlationKey) { - this(unmarked, Collections.> emptyList(), correlationKey, System.currentTimeMillis()); + public SimpleMessageGroup(Collection> unmarked, Object groupId) { + this(unmarked, Collections.> emptyList(), groupId, System.currentTimeMillis()); } public SimpleMessageGroup(Collection> unmarked, Collection> 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() { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java index 5831fd742b..c9bdae026c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java @@ -38,7 +38,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes private final ConcurrentMap> idToMessage; - private final ConcurrentMap correlationToMessageGroup; + private final ConcurrentMap 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>(); - this.correlationToMessageGroup = new ConcurrentHashMap(); + this.groupIdToMessageGroup = new ConcurrentHashMap(); 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 iterator() { - return new HashSet(correlationToMessageGroup.values()).iterator(); + return new HashSet(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); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTest.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTest.java index e4140e7c80..ae8926c667 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTest.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTest.java @@ -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(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreReaperTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreReaperTests.java index 1cca6c6e40..ce04db1c2a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreReaperTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreReaperTests.java @@ -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()); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java index 6bcf2a75e6..e4ea578bc5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java @@ -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; } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java index fd53672c7b..3878ce8c3b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java @@ -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()); } }); diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java index 3d87c07155..e5c678ad2f 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java @@ -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> marked = jdbcTemplate.query(getQuery(LIST_MARKED_MESSAGES_BY_CORRELATION_KEY), new Object[] { + public MessageGroup getMessageGroup(Object groupId) { + String key = getKey(groupId); + List> marked = jdbcTemplate.query(getQuery(LIST_MARKED_MESSAGES_BY_GROUP_KEY), new Object[] { key, region }, mapper); - List> unmarked = jdbcTemplate.query(getQuery(LIST_UNMARKED_MESSAGES_BY_CORRELATION_KEY), + List> 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 iterator() { @SuppressWarnings("unchecked") - final Iterator iterator = jdbcTemplate.query(getQuery(LIST_CORRELATION_KEYS), new Object[] { region }, + final Iterator iterator = jdbcTemplate.query(getQuery(LIST_GROUP_KEYS), new Object[] { region }, new SingleColumnRowMapper(String.class)).iterator(); return new Iterator() { diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql index 1624a139e3..fea32a78e3 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql @@ -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) ); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql index 1624a139e3..fea32a78e3 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql @@ -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) ); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql index 996df68083..42c89ff068 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql @@ -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) ); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql index 996df68083..42c89ff068 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql @@ -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) ); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql index 995cfb2ac4..1f751b690c 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql @@ -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) ); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle10g.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle10g.sql index 47ef6de35a..882d0ec9fb 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle10g.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle10g.sql @@ -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) ); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql index e8013dfe46..da2a30b079 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql @@ -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) ); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql index 7c8196be7e..6019a98e1d 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql @@ -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) ); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql index 7c8196be7e..6019a98e1d 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql @@ -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) ); diff --git a/spring-integration-jdbc/src/main/sql/schema.sql.vpp b/spring-integration-jdbc/src/main/sql/schema.sql.vpp index 541c637f4f..1e90a62b10 100644 --- a/spring-integration-jdbc/src/main/sql/schema.sql.vpp +++ b/spring-integration-jdbc/src/main/sql/schema.sql.vpp @@ -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) ); diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelTests.java index a5e8c2fde9..2f62fbe2a7 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelTests.java @@ -31,7 +31,7 @@ public class JdbcMessageStoreChannelTests { @Before public void clear() { for (MessageGroup group : messageStore) { - messageStore.removeMessageGroup(group.getCorrelationKey()); + messageStore.removeMessageGroup(group.getGroupId()); } } diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java index 84b6968bea..e7823a0ea6 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java @@ -112,11 +112,11 @@ public class JdbcMessageStoreTests { @Test @Transactional public void testAddAndGetMessageGroup() throws Exception { - String correlationId = "X"; - Message message = MessageBuilder.withPayload("foo").setCorrelationId(correlationId).build(); + String groupId = "X"; + Message 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 message = MessageBuilder.withPayload("foo").setCorrelationId(correlationId).build(); - messageStore.addMessageToGroup(correlationId, message); - messageStore.removeMessageFromGroup(correlationId, message); - MessageGroup group = messageStore.getMessageGroup(correlationId); + String groupId = "X"; + Message 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 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 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> 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 message = MessageBuilder.withPayload("foo").setCorrelationId(correlationId).build(); - messageStore.addMessageToGroup(correlationId, message); - MessageGroup group = messageStore.getMessageGroup(correlationId); + String groupId = "X"; + Message 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 message = MessageBuilder.withPayload("foo").setCorrelationId(correlationId).build(); - messageStore.addMessageToGroup(correlationId, message); + String groupId = "X"; + Message message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build(); + messageStore.addMessageToGroup(groupId, message); messageStore.registerMessageGroupExpiryCallback(new MessageGroupCallback() { public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { - messageGroupStore.removeMessageGroup(group.getCorrelationKey()); + messageGroupStore.removeMessageGroup(group.getGroupId()); } }); messageStore.expireMessageGroups(-10000); - MessageGroup group = messageStore.getMessageGroup(correlationId); + MessageGroup group = messageStore.getMessageGroup(groupId); assertEquals(0, group.size()); }