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 e61a23e04e..b57078ae29 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
@@ -1,11 +1,11 @@
/*
* Copyright 2002-2012 the original author or authors.
- *
+ *
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
* specific language governing permissions and limitations under the License.
@@ -23,7 +23,7 @@ import org.springframework.integration.Message;
/**
* 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
* @author Dave Syer
@@ -32,15 +32,15 @@ import org.springframework.integration.Message;
public class SimpleMessageGroup implements MessageGroup {
private final Object groupId;
-
+
public final BlockingQueue> messages = new LinkedBlockingQueue>();
-
+
private volatile int lastReleasedMessageSequence;
private final long timestamp;
-
+
private volatile long lastModified;
-
+
private volatile boolean complete;
public SimpleMessageGroup(Object groupId) {
@@ -56,22 +56,24 @@ public class SimpleMessageGroup implements MessageGroup {
this.timestamp = timestamp;
this.complete = complete;
for (Message> message : messages) {
- addMessage(message);
+ if (message != null){ //see INT-2666
+ addMessage(message);
+ }
}
}
public SimpleMessageGroup(MessageGroup messageGroup) {
this(messageGroup.getMessages(), messageGroup.getGroupId(), messageGroup.getTimestamp(), messageGroup.isComplete());
}
-
+
public long getTimestamp() {
return timestamp;
}
-
+
public void setLastModified(long lastModified){
this.lastModified = lastModified;
}
-
+
public long getLastModified() {
return lastModified;
}
@@ -87,7 +89,7 @@ public class SimpleMessageGroup implements MessageGroup {
public void remove(Message> message) {
messages.remove(message);
}
-
+
public int getLastReleasedMessageSequenceNumber() {
return lastReleasedMessageSequence;
}
@@ -99,7 +101,7 @@ public class SimpleMessageGroup implements MessageGroup {
public Collection> getMessages() {
return Collections.unmodifiableCollection(messages);
}
-
+
public void setLastReleasedMessageSequenceNumber(int sequenceNumber){
this.lastReleasedMessageSequence = sequenceNumber;
}
@@ -111,11 +113,11 @@ public class SimpleMessageGroup implements MessageGroup {
public boolean isComplete() {
return this.complete;
}
-
+
public void complete() {
this.complete = true;
}
-
+
public int getSequenceSize() {
if (size() == 0) {
return 0;
@@ -131,7 +133,7 @@ public class SimpleMessageGroup implements MessageGroup {
Message> one = messages.peek();
return one;
}
-
+
public void clear(){
this.messages.clear();
}
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageGroupTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageGroupTests.java
index c0b669f7e0..b072f6a419 100644
--- a/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageGroupTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageGroupTests.java
@@ -1,17 +1,20 @@
package org.springframework.integration.store;
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.mock;
+
import java.lang.reflect.Constructor;
+import java.util.ArrayList;
import java.util.Collections;
+import java.util.List;
import org.junit.Test;
import org.springframework.integration.Message;
import org.springframework.integration.support.MessageBuilder;
-import static org.hamcrest.CoreMatchers.is;
-
-import static org.junit.Assert.assertThat;
-
/**
* @author Iwein Fuld
* @author Oleg Zhurakousky
@@ -19,7 +22,7 @@ import static org.junit.Assert.assertThat;
*/
public class SimpleMessageGroupTests {
- private Object key = new Object();
+ private final Object key = new Object();
private SimpleMessageGroup group = new SimpleMessageGroup(Collections.> emptyList(), key);
@@ -53,4 +56,16 @@ public class SimpleMessageGroupTests {
group.add(message2);
assertThat(group.canAdd(message1), is(true));
}
+
+ @Test // shoudl not fail with NPE (see INT-2666)
+ public void shouldIgnoreNullValuesWhenInitializedWithCollectionContainingNulls() throws Exception{
+ Message> m1 = mock(Message.class);
+ Message> m2 = mock(Message.class);
+ final List> messages = new ArrayList>();
+ messages.add(m1);
+ messages.add(null);
+ messages.add(m2);
+ SimpleMessageGroup grp = new SimpleMessageGroup(messages, 1);
+ assertEquals(2, grp.getMessages().size());
+ }
}