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

Merge pull request #162 from olegz/INT-2212

Browse Source 
fixed NPE in AbstractKeyValueMessageStore
This commit is contained in:
Mark Fisher
2011-11-01 17:07:53 -04:00
parent 61b7108f72 2d504904db
commit bcc17ee497
5 changed files with 77 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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;
}

17
spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java
View File

@@ -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<Long, UUID> 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<Long, UUID> firstEntry = messageCreationDateToIdMappings.firstEntry();
if (firstEntry != null){
return firstEntry.getValue();
}
return null;
}
public boolean isComplete() {

3
spring-integration-gemfire/.springBeans
View File

@@ -1,7 +1,7 @@
<?xml version="1.0" encoding="UTF-8"?>
<beansProjectDescription>
<version>1</version>
<pluginVersion><![CDATA[2.8.0.201108100015-M1]]></pluginVersion>
<pluginVersion><![CDATA[2.8.0.201110170010-RELEASE]]></pluginVersion>
<configSuffixes>
<configSuffix><![CDATA[xml]]></configSuffix>
</configSuffixes>
@@ -9,6 +9,7 @@
<configs>
<config>src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests-context.xml</config>
<config>src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests-context.xml</config>
<config>src/test/java/org/springframework/integration/gemfire/store/gemfire-queue-context.xml</config>
</configs>
<configSets>
</configSets>

16
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java
View File

@@ -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<String>("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{

27
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/gemfire-queue-config.xml Normal file
View File

@@ -0,0 +1,27 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.1.xsd">
<int:channel id="gemfireQueue">
<int:queue message-store="gemfireStore"/>
</int:channel>
<int:bridge input-channel="gemfireQueue" output-channel="outputQueue">
<int:poller fixed-rate="1000" max-messages-per-poll="200"/>
</int:bridge>
<int:channel id="outputQueue">
<int:queue/>
</int:channel>
<bean id="gemfireStore" class="org.springframework.integration.gemfire.store.GemfireMessageStore">
<constructor-arg ref="gemfireCache"/>
</bean>
<bean id="gemfireCache" class="org.springframework.data.gemfire.CacheFactoryBean"/>
</beans>