Added MessageStoringInterceptor (INT-264).

org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/MessageStoringInterceptor.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2002-2008 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.
+ */
+
+package org.springframework.integration.channel.interceptor;
+
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.integration.channel.ChannelInterceptor;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageStore;
+import org.springframework.integration.message.MessagingException;
+import org.springframework.util.Assert;
+
+/**
+ * A {@link ChannelInterceptor} that delegates to a {@link MessageStore}.
+ * Upon sending to the channel, the message will be added to the store,
+ * and upon receiving from the channel, the message will be removed from
+ * the store.
+ * 
+ * @author Mark Fisher
+ */
+public class MessageStoringInterceptor extends ChannelInterceptorAdapter {
+
+	private final Log logger = LogFactory.getLog(this.getClass());
+
+	private final MessageStore messageStore;
+
+	private volatile boolean initialized;
+
+	private final Object initializationMonitor = new Object();
+
+
+	public MessageStoringInterceptor(MessageStore messageStore) {
+		Assert.notNull(messageStore, "MessageStore must not be null");
+		this.messageStore = messageStore;
+	}
+
+
+	@Override
+	public boolean preSend(Message<?> message, MessageChannel channel) {
+		if (message != null) {
+			this.messageStore.put(message.getId(), message);
+		}
+		return true;
+	}
+
+	@Override
+	public boolean preReceive(MessageChannel channel) {
+		synchronized (this.initializationMonitor) {
+			if (!this.initialized) {
+				if (logger.isDebugEnabled()) {
+					logger.debug("pre-loading Messages from MessageStore for channel '" + channel.getName() + "'");
+				}
+				List<Message<?>> storedMessages = this.messageStore.list();
+				for (Message<?> message : storedMessages) {
+					if (!channel.send(message, 0)) {
+						throw new MessagingException("failed to initialize channel from MessageStore");
+					}
+				}
+				if (logger.isDebugEnabled()) {
+					logger.debug("pre-loaded " + storedMessages.size() + " Messages for channel '" + channel.getName() + "'");
+				}
+			}
+			this.initialized = true;
+		}
+		return true;
+	}
+
+	@Override
+	public void postReceive(Message<?> message, MessageChannel channel) {
+		if (message != null) {
+			this.messageStore.remove(message.getId());
+		}
+	}
+
+}

org.springframework.integration/src/main/java/org/springframework/integration/message/MessageStore.java

@@ -16,6 +16,8 @@
 
 package org.springframework.integration.message;
 
+import java.util.List;
+
 /**
  * Strategy interface for storing and retrieving messages.
  * 
@@ -29,6 +31,8 @@ public interface MessageStore {
 
 	Message<?> remove(Object key);
 
+	List<Message<?>> list();
+
 	int size();
 
 }

org.springframework.integration/src/main/java/org/springframework/integration/message/RetrievalBlockingMessageStore.java

@@ -102,6 +102,10 @@ public class RetrievalBlockingMessageStore implements MessageStore {
 		return (message != null) ? message : waitForMessage(key, timeout, false);
 	}
 
+	public List<Message<?>> list() {
+		return this.targetMessageStore.list();
+	}
+
 	public Message<?> remove(Object key) {
 		return this.remove(key, -1);
 	}

org.springframework.integration/src/main/java/org/springframework/integration/message/SimpleMessageStore.java

@@ -16,6 +16,8 @@
 
 package org.springframework.integration.message;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 
 import org.springframework.integration.util.BoundedHashMap;
@@ -46,6 +48,10 @@ public class SimpleMessageStore implements MessageStore {
 		return (key != null) ? this.map.get(key) : null;
 	}
 
+	public List<Message<?>> list() {
+		return new ArrayList<Message<?>>(this.map.values());
+	}
+
 	public Message<?> remove(Object key) {
 		return (key != null) ? this.map.remove(key) : null;
 	}

org.springframework.integration/src/test/java/org/springframework/integration/message/SimpleMessageStoreTests.java

@@ -20,6 +20,8 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 
+import java.util.List;
+
 import org.junit.Test;
 
 /**
@@ -79,4 +81,33 @@ public class SimpleMessageStoreTests {
 		assertEquals(message4, store.get(4));
 	}
 
+	@Test
+	public void testListWhenEmpty() {
+		SimpleMessageStore store = new SimpleMessageStore(3);
+		List<Message<?>> list = store.list();
+		assertNotNull(list);
+		assertEquals(0, list.size());
+	}
+
+	@Test
+	public void testListWhenUnderCapacity() {
+		SimpleMessageStore store = new SimpleMessageStore(3);
+		store.put(1, new StringMessage("foo"));
+		store.put(2, new StringMessage("bar"));
+		List<Message<?>> list = store.list();
+		assertEquals(2, list.size());
+	}
+
+	@Test
+	public void testListAfterExceedingCapacity() {
+		SimpleMessageStore store = new SimpleMessageStore(2);
+		store.put(1, new StringMessage("foo"));
+		store.put(2, new StringMessage("bar"));
+		store.put(3, new StringMessage("baz"));
+		List<Message<?>> list = store.list();
+		assertEquals(2, list.size());
+		assertEquals("bar", list.get(0).getPayload());
+		assertEquals("baz", list.get(1).getPayload());
+	}
+
 }