INT-2666 Fix NPE in SimpleMessageStore

INT-2666 polished test

Polishing

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 <code>add</code>ed 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<Message<?>> messages = new LinkedBlockingQueue<Message<?>>();
-	
+
 	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<Message<?>> 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();
 	}