From 2d504904db78c4c6d08a8f3b275df6b3507cbf0f Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Mon, 31 Oct 2011 16:55:28 -0400 Subject: [PATCH] INT-2212 fixed NPE in AbstractKeyValueMessageStore INT-2212 polished AbstractKeyValueMessageStore eliminated possibility for NPE during the Message removal, added test INT-2212 polishing INT-2212 polishing added remove(..) method to MessageGroupMetadata --- .../store/AbstractKeyValueMessageStore.java | 25 ++++++++++------- .../store/MessageGroupMetadata.java | 17 +++++++++++- spring-integration-gemfire/.springBeans | 3 ++- .../gemfire/store/GemfireGroupStoreTests.java | 16 +++++++++++ .../gemfire/store/gemfire-queue-config.xml | 27 +++++++++++++++++++ 5 files changed, 77 insertions(+), 11 deletions(-) create mode 100644 spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/gemfire-queue-config.xml diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java index fbf0fbd105..493a58377e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java @@ -42,8 +42,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS protected static final String MESSAGE_GROUP_KEY_PREFIX = "MESSAGE_GROUP_"; protected static final String CREATED_DATE = "CREATED_DATE"; - - + // MessageStore methods public Message getMessage(UUID id) { @@ -68,7 +67,10 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS if (message != null) { Assert.isInstanceOf(Message.class, message); } - return (Message) message; + if (message != null){ + return this.normalizeMessage((Message) message); + } + return null; } @ManagedAttribute @@ -146,7 +148,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS public void completeGroup(Object groupId) { Assert.notNull(groupId, "'groupId' must not be null"); - SimpleMessageGroup messageGroup = this.buildMessageGroup(this.getMessageGroup(groupId), true); + SimpleMessageGroup messageGroup = this.buildMessageGroup(groupId, true); messageGroup.complete(); this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup)); } @@ -182,9 +184,15 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS Assert.isInstanceOf(MessageGroupMetadata.class, mgm); MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm; - Message message = this.removeMessage(messageGroupMetadata.firstId()); - Message normalizedMessage = this.normalizeMessage(message); - return normalizedMessage; + UUID firstId = messageGroupMetadata.firstId(); + if (firstId != null){ + Message message = this.removeMessage(firstId); + + messageGroupMetadata.remove(message.getHeaders().getId()); + this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, messageGroupMetadata); + + return message; + } } return null; } @@ -273,8 +281,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS private Message getRawMessage(UUID id) { Assert.notNull(id, "'id' must not be null"); - Object message = this.doRetrieve(MESSAGE_KEY_PREFIX + id); - Assert.isInstanceOf(Message.class, message); + Object message = this.doRetrieve(MESSAGE_KEY_PREFIX + id); return (Message) message; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java index 469dad2b77..91d099fe56 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java @@ -18,6 +18,7 @@ package org.springframework.integration.store; import java.io.Serializable; import java.util.Iterator; +import java.util.Map.Entry; import java.util.TreeMap; import java.util.UUID; @@ -62,6 +63,16 @@ public class MessageGroupMetadata implements Serializable{ this.lastReleasedMessageSequenceNumber = messageGroup.getLastReleasedMessageSequenceNumber(); } + public void remove(UUID messageId){ + long currentTimestamp = 0; + for (Entry entry : messageCreationDateToIdMappings.entrySet()) { + if (entry.getValue().equals(messageId)){ + currentTimestamp = entry.getKey(); + break; + } + } + this.messageCreationDateToIdMappings.remove(currentTimestamp); + } public Object getGroupId() { return this.groupId; @@ -72,7 +83,11 @@ public class MessageGroupMetadata implements Serializable{ } public UUID firstId(){ - return messageCreationDateToIdMappings.firstEntry().getValue(); + Entry firstEntry = messageCreationDateToIdMappings.firstEntry(); + if (firstEntry != null){ + return firstEntry.getValue(); + } + return null; } public boolean isComplete() { diff --git a/spring-integration-gemfire/.springBeans b/spring-integration-gemfire/.springBeans index 90b1f9cb62..4b08dc36b5 100644 --- a/spring-integration-gemfire/.springBeans +++ b/spring-integration-gemfire/.springBeans @@ -1,7 +1,7 @@ 1 - + @@ -9,6 +9,7 @@ src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests-context.xml src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests-context.xml + src/test/java/org/springframework/integration/gemfire/store/gemfire-queue-context.xml diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java index 974776c185..52aa45d341 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java @@ -296,6 +296,22 @@ public class GemfireGroupStoreTests { inputA.send(m3); assertNotNull(outputA.receive(1000)); } + @Test + public void testQueue() throws Exception{ + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("gemfire-queue-config.xml", this.getClass()); + + QueueChannel gemfireQueue = context.getBean("gemfireQueue", QueueChannel.class); + QueueChannel outputQueue = context.getBean("outputQueue", QueueChannel.class); + + for (int i = 0; i < 20; i++) { + gemfireQueue.send(new GenericMessage("Hello")); + Thread.sleep(1); + } + for (int i = 0; i < 20; i++) { + assertNotNull(outputQueue.receive(1)); + } + assertNull(outputQueue.receive(1)); + } @Before public void init() throws Exception{ diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/gemfire-queue-config.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/gemfire-queue-config.xml new file mode 100644 index 0000000000..2173ea5674 --- /dev/null +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/gemfire-queue-config.xml @@ -0,0 +1,27 @@ + + + + + + + + + + + + + + + + + + + + + + +