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

INT-1069: Change MessageGroup interfaces to better match use cases

Browse Source
This commit is contained in:
David Syer
2010-06-18 10:49:43 +00:00
parent d5614ff3e8
commit 4e4cb559d1
17 changed files with 127 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-integration-core/.classpath
View File

@@ -4,6 +4,10 @@
<classpathentry kind="src" output="target/classes" path="src/main/resources"/>
<classpathentry kind="src" output="target/test-classes" path="src/test/java"/>
<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER"/>
<classpathentry kind="con" path="org.maven.ide.eclipse.MAVEN2_CLASSPATH_CONTAINER"/>
<classpathentry kind="con" path="org.maven.ide.eclipse.MAVEN2_CLASSPATH_CONTAINER">
<attributes>
<attribute name="org.eclipse.jst.component.nondependency" value=""/>
</attributes>
</classpathentry>
<classpathentry kind="output" path="target/classes"/>
</classpath>

2
spring-integration-core/.settings/org.eclipse.wst.common.component
View File

@@ -3,7 +3,5 @@
<wb-module deploy-name="spring-integration-core">
<wb-resource deploy-path="/" source-path="/src/main/java"/>
<wb-resource deploy-path="/" source-path="/src/main/resources"/>
<wb-resource deploy-path="/" source-path="/src/test/resources"/>
<wb-resource deploy-path="/" source-path="/src/test/java"/>
</wb-module>
</project-modules>

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

@@ -99,7 +99,7 @@ public class CorrelatingMessageBarrier extends AbstractMessageHandler implements
Iterator<Message<?>> unmarked = group.getUnmarked().iterator();
if (unmarked.hasNext()) {
nextMessage = unmarked.next();
store.markMessageInGroup(key, nextMessage);
store.removeMessageFromGroup(key, nextMessage);
if (log.isDebugEnabled()) {
log.debug(String.format("Released message for key [%s]: %s.", key, nextMessage));
}

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

@@ -165,9 +165,9 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
MessageGroup group = messageStore.getMessageGroup(correlationKey);
if (group.add(message)) {
if (group.canAdd(message)) {
store(correlationKey, message);
group = store(correlationKey, message);
if (releaseStrategy.canRelease(group)) {
@@ -281,8 +281,8 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
locks.remove(correlationKey);
}
private void store(Object correlationKey, Message<?> message) {
messageStore.addMessageToGroup(correlationKey, message);
private MessageGroup store(Object correlationKey, Message<?> message) {
return messageStore.addMessageToGroup(correlationKey, message);
}
}

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

@@ -15,10 +15,9 @@ import java.util.Collection;
public interface MessageGroup {
/**
* Add a message to the internal list. This is needed to avoid hitting the
* underlying store or copying the internal list.
* Query if the message can be added.
*/
boolean add(Message<?> message);
boolean canAdd(Message<?> message);
/**
* @return unmarked messages in the group at time of the invocation

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

@@ -39,7 +39,7 @@ public interface MessageGroupStore {
* @param correlationKey the correlation id to store the message under
* @param message a message
*/
void addMessageToGroup(Object correlationKey, Message<?> message);
MessageGroup addMessageToGroup(Object correlationKey, Message<?> message);
/**
* Persist the mark on all the messages from the group. The group is modified in the process as all its unmarked
@@ -47,7 +47,7 @@ public interface MessageGroupStore {
*
* @param group a MessageGroup with no unmarked messages
*/
void markMessageGroup(MessageGroup group);
MessageGroup markMessageGroup(MessageGroup group);
/**
* Persist a mark on a single message from the group. The group is modified to reflect that 'messageToMark' is no
@@ -55,7 +55,7 @@ public interface MessageGroupStore {
* @param key the correlationKey for the group containing the message
* @param messageToMark the message to be marked
*/
void markMessageInGroup(Object key, Message<?> messageToMark);
MessageGroup removeMessageFromGroup(Object key, Message<?> messageToMark);
/**
* Remove the message group with this correlation key.

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

@@ -13,18 +13,18 @@
package org.springframework.integration.store;
import org.springframework.integration.core.Message;
import java.util.Collection;
import java.util.Collections;
import java.util.concurrent.BlockingQueue;
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.
*
*
* @author Iwein Fuld
* @author Oleg Zhurakousky
* @author Dave Syer
@@ -34,26 +34,28 @@ public class SimpleMessageGroup implements MessageGroup {
private final Object correlationKey;
//Guards(marked, unmarked)
// Guards(marked, unmarked)
private final Object lock = new Object();
//@GuardedBy(lock)
// @GuardedBy(lock)
public final BlockingQueue<Message<?>> marked = new LinkedBlockingQueue<Message<?>>();
//@GuardedBy(lock)
// @GuardedBy(lock)
public final BlockingQueue<Message<?>> unmarked = new LinkedBlockingQueue<Message<?>>();
private final long timestamp;
public SimpleMessageGroup(Object correlationKey) {
this(Collections.<Message<?>>emptyList(), Collections.<Message<?>>emptyList(), correlationKey, System.currentTimeMillis());
this(Collections.<Message<?>> emptyList(), Collections.<Message<?>> emptyList(), correlationKey, System
.currentTimeMillis());
}
public SimpleMessageGroup(Collection<? extends Message<?>> unmarked, Object correlationKey) {
this(unmarked, Collections.<Message<?>>emptyList(), correlationKey, System.currentTimeMillis());
this(unmarked, Collections.<Message<?>> emptyList(), correlationKey, System.currentTimeMillis());
}
public SimpleMessageGroup(Collection<? extends Message<?>> unmarked,
Collection<? extends Message<?>> marked,
Object correlationKey, long timestamp) {
public SimpleMessageGroup(Collection<? extends Message<?>> unmarked, Collection<? extends Message<?>> marked,
Object correlationKey, long timestamp) {
this.correlationKey = correlationKey;
this.timestamp = timestamp;
synchronized (lock) {
@@ -79,8 +81,19 @@ public class SimpleMessageGroup implements MessageGroup {
return timestamp;
}
public boolean add(Message<?> message) {
return addUnmarked(message);
public boolean canAdd(Message<?> message) {
return !isMember(message);
}
public void add(Message<?> message) {
addUnmarked(message);
}
public void remove(Message<?> message) {
synchronized (lock) {
marked.remove(message);
unmarked.remove(message);
}
}
private boolean addUnmarked(Message<?> message) {
@@ -170,7 +183,8 @@ public class SimpleMessageGroup implements MessageGroup {
Integer messageSequenceSize = message.getHeaders().getSequenceSize();
if (!messageSequenceSize.equals(getSequenceSize())) {
return true;
} else {
}
else {
synchronized (lock) {
if (containsSequenceNumber(unmarked, messageSequenceNumber)
|| containsSequenceNumber(marked, messageSequenceNumber)) {
@@ -191,4 +205,5 @@ public class SimpleMessageGroup implements MessageGroup {
}
return false;
}
}

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

@@ -13,17 +13,17 @@
package org.springframework.integration.store;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.util.UpperBound;
import org.springframework.util.Assert;
import java.util.HashSet;
import java.util.Iterator;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.util.UpperBound;
import org.springframework.util.Assert;
/**
* Map-based implementation of {@link MessageStore} and {@link MessageGroupStore}. Enforces a maximum capacity for the
* store.
@@ -89,7 +89,8 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes
if (key != null) {
individualUpperBound.release();
return this.idToMessage.remove(key);
} else
}
else
return null;
}
@@ -102,19 +103,22 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes
return new SimpleMessageGroup(group);
}
public void addMessageToGroup(Object correlationId, Message<?> message) {
public MessageGroup addMessageToGroup(Object correlationId, Message<?> message) {
if (!groupUpperBound.tryAcquire(0)) {
throw new MessagingException(this.getClass().getSimpleName()
+ " was out of capacity at, try constructing it with a larger capacity.");
}
getMessageGroupInternal(correlationId).add(message);
SimpleMessageGroup group = getMessageGroupInternal(correlationId);
group.add(message);
return group;
}
public void markMessageGroup(MessageGroup group) {
public MessageGroup markMessageGroup(MessageGroup group) {
Object correlationId = group.getCorrelationKey();
MessageGroup internal = getMessageGroupInternal(correlationId);
internal.markAll();
group.markAll();
return group;
}
public void removeMessageGroup(Object correlationId) {
@@ -122,8 +126,10 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes
correlationToMessageGroup.remove(correlationId);
}
public void markMessageInGroup(Object key, Message<?> messageToMark) {
getMessageGroupInternal(key).mark(messageToMark);
public MessageGroup removeMessageFromGroup(Object key, Message<?> messageToMark) {
SimpleMessageGroup group = getMessageGroupInternal(key);
group.remove(messageToMark);
return group;
}
@Override

4
spring-integration-core/src/test/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategyTests.java
View File

@@ -34,7 +34,7 @@ public class SequenceSizeReleaseStrategyTests {
public void testIncompleteList() {
Message<String> message = MessageBuilder.withPayload("test1")
.setSequenceSize(2).build();
MessageGroup messages = new SimpleMessageGroup("FOO");
SimpleMessageGroup messages = new SimpleMessageGroup("FOO");
messages.add(message);
SequenceSizeReleaseStrategy releaseStrategy = new SequenceSizeReleaseStrategy();
assertFalse(releaseStrategy.canRelease(messages));
@@ -46,7 +46,7 @@ public class SequenceSizeReleaseStrategyTests {
.setSequenceSize(2).build();
Message<String> message2 = MessageBuilder.withPayload("test2")
.setSequenceSize(2).build();
MessageGroup messages = new SimpleMessageGroup("FOO");
SimpleMessageGroup messages = new SimpleMessageGroup("FOO");
messages.add(message1);
messages.add(message2);
SequenceSizeReleaseStrategy releaseStrategy = new SequenceSizeReleaseStrategy();

8
spring-integration-core/src/test/java/org/springframework/integration/aggregator/TimeoutCountSequenceSizeReleaseStrategyTests.java
View File

@@ -34,7 +34,7 @@ public class TimeoutCountSequenceSizeReleaseStrategyTests {
public void testIncompleteList() {
Message<String> message = MessageBuilder.withPayload("test1")
.setSequenceSize(2).build();
MessageGroup messages = new SimpleMessageGroup("FOO");
SimpleMessageGroup messages = new SimpleMessageGroup("FOO");
messages.add(message);
TimeoutCountSequenceSizeReleaseStrategy releaseStrategy = new TimeoutCountSequenceSizeReleaseStrategy();
assertFalse(releaseStrategy.canRelease(messages));
@@ -44,7 +44,7 @@ public class TimeoutCountSequenceSizeReleaseStrategyTests {
public void testIncompleteListWithTimeout() {
Message<String> message = MessageBuilder.withPayload("test1")
.setSequenceSize(2).build();
MessageGroup messages = new SimpleMessageGroup("FOO");
SimpleMessageGroup messages = new SimpleMessageGroup("FOO");
messages.add(message);
TimeoutCountSequenceSizeReleaseStrategy releaseStrategy = new TimeoutCountSequenceSizeReleaseStrategy(TimeoutCountSequenceSizeReleaseStrategy.DEFAULT_THRESHOLD, -100);
assertTrue(releaseStrategy.canRelease(messages));
@@ -54,7 +54,7 @@ public class TimeoutCountSequenceSizeReleaseStrategyTests {
public void testIncompleteListWithCount() {
Message<String> message = MessageBuilder.withPayload("test1")
.setSequenceSize(2).build();
MessageGroup messages = new SimpleMessageGroup("FOO");
SimpleMessageGroup messages = new SimpleMessageGroup("FOO");
messages.add(message);
TimeoutCountSequenceSizeReleaseStrategy releaseStrategy = new TimeoutCountSequenceSizeReleaseStrategy(1, TimeoutCountSequenceSizeReleaseStrategy.DEFAULT_TIMEOUT);
assertTrue(releaseStrategy.canRelease(messages));
@@ -66,7 +66,7 @@ public class TimeoutCountSequenceSizeReleaseStrategyTests {
.setSequenceSize(2).build();
Message<String> message2 = MessageBuilder.withPayload("test2")
.setSequenceSize(2).build();
MessageGroup messages = new SimpleMessageGroup("FOO");
SimpleMessageGroup messages = new SimpleMessageGroup("FOO");
messages.add(message1);
messages.add(message2);
SequenceSizeReleaseStrategy releaseStrategy = new SequenceSizeReleaseStrategy();

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

@@ -16,15 +16,19 @@
package org.springframework.integration.store;
import static org.junit.Assert.assertEquals;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import org.junit.Test;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.StringMessage;
import org.springframework.test.util.ReflectionTestUtils;
import java.util.*;
import static org.junit.Assert.assertEquals;
/**
* @author Dave Syer
*/
@@ -68,17 +72,20 @@ public class MessageStoreTests {
return Arrays.asList(testMessages).iterator();
}
public void addMessageToGroup(Object correlationKey, Message<?> message) {
public MessageGroup addMessageToGroup(Object correlationKey, Message<?> message) {
throw new UnsupportedOperationException();
}
public MessageGroup getMessageGroup(Object correlationKey) {
return removed ? new SimpleMessageGroup(correlationKey) : testMessages;
}
public void markMessageGroup(MessageGroup group) {
public MessageGroup markMessageGroup(MessageGroup group) {
throw new UnsupportedOperationException();
}
public void markMessageInGroup(Object key, Message<?> messageToMark) {
public MessageGroup removeMessageFromGroup(Object key, Message<?> messageToMark) {
throw new UnsupportedOperationException();
}
public void removeMessageGroup(Object correlationKey) {

19
spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageGroupTests.java
View File

@@ -5,11 +5,9 @@ import static org.junit.Assert.assertThat;
import java.util.Collections;
import org.junit.Before;
import org.junit.Test;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.store.SimpleMessageGroup;
/**
* @author Iwein Fuld
@@ -20,28 +18,25 @@ public class SimpleMessageGroupTests {
private Object key = new Object();
private MessageGroup group;
@Before
public void buildMessageGroup() {
group = new SimpleMessageGroup(Collections.<Message<?>> emptyList(), key);
}
private SimpleMessageGroup group = new SimpleMessageGroup(Collections.<Message<?>> emptyList(), key);
@Test
public void shouldFindSupersedingMessages() {
final Message<?> message1 = MessageBuilder.withPayload("test").setSequenceNumber(1).build();
final Message<?> message2 = MessageBuilder.fromMessage(message1).setSequenceNumber(1).build();
assertThat(group.add(message1), is(true));
assertThat(group.canAdd(message1), is(true));
group.add(message1);
group.add(message2);
assertThat(group.add(message1), is(false));
assertThat(group.canAdd(message1), is(false));
}
@Test
public void shouldIgnoreMessagesWithZeroSequenceNumber() {
final Message<?> message1 = MessageBuilder.withPayload("test").build();
final Message<?> message2 = MessageBuilder.fromMessage(message1).build();
assertThat(group.add(message1), is(true));
assertThat(group.canAdd(message1), is(true));
group.add(message1);
group.add(message2);
assertThat(group.add(message1), is(true));
assertThat(group.canAdd(message1), is(true));
}
}

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

@@ -82,6 +82,17 @@ public class SimpleMessageStoreTests {
assertEquals(1, store.getMessageGroup("bar").size());
}
@Test
public void shouldRemoveFromGroup() throws Exception {
SimpleMessageStore store = new SimpleMessageStore();
Message<String> testMessage1 = MessageBuilder.withPayload("foo").build();
store.addMessageToGroup("bar", testMessage1);
Message<?> testMessage2 = store.getMessageGroup("bar").getOne();
MessageGroup group = store.removeMessageFromGroup("bar", testMessage2);
assertEquals(0, group.size());
assertEquals(0, store.getMessageGroup("bar").size());
}
@Test
public void shouldCopyMessageGroup() throws Exception {
SimpleMessageStore store = new SimpleMessageStore();

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

@@ -76,7 +76,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
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_MESSAGE_IN_GROUP = "UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=?, MARKED=1 where MARKED=0 and CORRELATION_KEY=? and REGION=? and MESSAGE_ID=?";
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 DELETE_MESSAGE_GROUP = "DELETE from %PREFIX%MESSAGE_GROUP where CORRELATION_KEY=? and REGION=?";
@@ -242,7 +242,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
return result;
}
public void addMessageToGroup(Object correlationKey, Message<?> message) {
public MessageGroup addMessageToGroup(Object correlationKey, Message<?> message) {
final long createdDate = System.currentTimeMillis();
final String messageId = getKey(message.getHeaders().getId());
@@ -259,6 +259,8 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
lobHandler.getLobCreator().setBlobAsBytes(ps, 5, messageBytes);
}
});
return getMessageGroup(correlationKey);
}
@@ -278,7 +280,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
return new SimpleMessageGroup(unmarked, marked, correlationKey, timestamp);
}
public void markMessageGroup(MessageGroup group) {
public MessageGroup markMessageGroup(MessageGroup group) {
final long updatedDate = System.currentTimeMillis();
final String correlationId = getKey(group.getCorrelationKey());
@@ -293,15 +295,17 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
});
group.markAll();
return group;
}
public void markMessageInGroup(Object key, Message<?> messageToMark) {
public MessageGroup removeMessageFromGroup(Object correlationKey, Message<?> messageToMark) {
final long updatedDate = System.currentTimeMillis();
final String correlationId = getKey(key);
final String correlationId = getKey(correlationKey);
final String messageId = getKey(messageToMark.getHeaders().getId());
jdbcTemplate.update(getQuery(MARK_MESSAGE_IN_GROUP), new PreparedStatementSetter() {
jdbcTemplate.update(getQuery(REMOVE_MESSAGE_FROM_GROUP), new PreparedStatementSetter() {
public void setValues(PreparedStatement ps) throws SQLException {
logger.debug("Marking messages with correlation key=" + correlationId);
ps.setTimestamp(1, new Timestamp(updatedDate));
@@ -310,6 +314,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
ps.setString(4, messageId);
}
});
return getMessageGroup(correlationKey);
}
public void removeMessageGroup(Object correlationKey) {

6
spring-integration-samples/ws-inbound-gateway/.settings/org.eclipse.wst.common.component
View File

@@ -3,6 +3,12 @@
<wb-module deploy-name="ws-inbound-gateway">
<wb-resource deploy-path="/WEB-INF/classes" source-path="/src/main/java"/>
<wb-resource deploy-path="/" source-path="/src/main/webapp"/>
<dependent-module deploy-path="/WEB-INF/lib" handle="module:/resource/spring-integration-xml/spring-integration-xml">
<dependency-type>uses</dependency-type>
</dependent-module>
<dependent-module deploy-path="/WEB-INF/lib" handle="module:/resource/spring-integration-core/spring-integration-core">
<dependency-type>uses</dependency-type>
</dependent-module>
<property name="java-output-path" value="/ws-inbound-gateway/target/classes"/>
<property name="context-root" value="ws-inbound-gateway"/>
</wb-module>

6
spring-integration-xml/.classpath
View File

@@ -4,6 +4,10 @@
<classpathentry kind="src" output="target/classes" path="src/main/resources"/>
<classpathentry kind="src" output="target/test-classes" path="src/test/java"/>
<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER"/>
<classpathentry kind="con" path="org.maven.ide.eclipse.MAVEN2_CLASSPATH_CONTAINER"/>
<classpathentry kind="con" path="org.maven.ide.eclipse.MAVEN2_CLASSPATH_CONTAINER">
<attributes>
<attribute name="org.eclipse.jst.component.nondependency" value=""/>
</attributes>
</classpathentry>
<classpathentry kind="output" path="target/classes"/>
</classpath>

6
spring-integration-xml/.project
View File

@@ -5,6 +5,11 @@
<projects>
</projects>
<buildSpec>
<buildCommand>
<name>org.eclipse.wst.common.project.facet.core.builder</name>
<arguments>
</arguments>
</buildCommand>
<buildCommand>
<name>org.eclipse.jdt.core.javabuilder</name>
<arguments>
@@ -19,5 +24,6 @@
<natures>
<nature>org.maven.ide.eclipse.maven2Nature</nature>
<nature>org.eclipse.jdt.core.javanature</nature>
<nature>org.eclipse.wst.common.project.facet.core.nature</nature>
</natures>
</projectDescription>