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

INT-1847: modify SimpleMessageStore to behave correctly with a finite capacity

Browse Source
This commit is contained in:
Dave Syer
2011-03-24 08:13:58 +00:00
parent 6e58a9ff99
commit b81f8b12ee
3 changed files with 31 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -74,7 +74,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes
public SimpleMessageStore() {
this(0);
}
@ManagedAttribute
public int getMessageCount() {
return idToMessage.size();
@@ -101,7 +101,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes
else
return null;
}
public MessageGroup getMessageGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
SimpleMessageGroup group = groupIdToMessageGroup.get(groupId);
@@ -129,7 +129,10 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes
}
public void removeMessageGroup(Object groupId) {
groupUpperBound.release();
if (!groupIdToMessageGroup.containsKey(groupId)) {
return;
}
groupUpperBound.release(groupIdToMessageGroup.get(groupId).size());
groupIdToMessageGroup.remove(groupId);
}

10
spring-integration-core/src/main/java/org/springframework/integration/util/UpperBound.java
View File

@@ -73,4 +73,14 @@ public final class UpperBound {
}
}
/**
* Releases several locks on the underlying semaphore. This is typically not done by the same Thread
* that acquired the lock, but by the thread that picked up the message.
*/
public void release(int permits) {
if (this.semaphore != null) {
this.semaphore.release(permits);
}
}
}

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

@@ -73,7 +73,7 @@ public class SimpleMessageStoreTests {
store.addMessage(testMessage1);
store.addMessage(testMessage2);
}
@Test
public void shouldListByCorrelation() throws Exception {
SimpleMessageStore store = new SimpleMessageStore();
@@ -93,6 +93,18 @@ public class SimpleMessageStoreTests {
assertEquals(0, store.getMessageGroup("bar").size());
}
@Test
public void testRepeatedAddAndRemoveGroup() throws Exception {
SimpleMessageStore store = new SimpleMessageStore(10, 10);
for (int i = 0; i < 10; i++) {
store.addMessageToGroup("bar", MessageBuilder.withPayload("foo").build());
store.addMessageToGroup("bar", MessageBuilder.withPayload("foo").build());
store.removeMessageGroup("bar");
assertEquals(0, store.getMessageGroup("bar").size());
assertEquals(0, store.getMessageGroupCount());
}
}
@Test
public void shouldCopyMessageGroup() throws Exception {
SimpleMessageStore store = new SimpleMessageStore();
@@ -104,11 +116,11 @@ public class SimpleMessageStoreTests {
@Test
public void shouldRegisterCallbacks() throws Exception {
SimpleMessageStore store = new SimpleMessageStore();
store.setExpiryCallbacks(Arrays.<MessageGroupCallback>asList(new MessageGroupCallback() {
store.setExpiryCallbacks(Arrays.<MessageGroupCallback> asList(new MessageGroupCallback() {
public void execute(MessageGroupStore messageGroupStore, MessageGroup group) {
}
}));
assertEquals(1, ((Collection<?>)ReflectionTestUtils.getField(store, "expiryCallbacks")).size());
assertEquals(1, ((Collection<?>) ReflectionTestUtils.getField(store, "expiryCallbacks")).size());
}
@Test