diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/MessageStoringInterceptor.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/MessageStoringInterceptor.java new file mode 100644 index 0000000000..f3522615ae --- /dev/null +++ b/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> 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()); + } + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageStore.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageStore.java index dc14b84a39..7b9f281131 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageStore.java +++ b/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> list(); + int size(); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/RetrievalBlockingMessageStore.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/RetrievalBlockingMessageStore.java index a0a60b5d93..738a09cb7e 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/RetrievalBlockingMessageStore.java +++ b/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> list() { + return this.targetMessageStore.list(); + } + public Message remove(Object key) { return this.remove(key, -1); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/SimpleMessageStore.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/SimpleMessageStore.java index fbe10830d3..2a335952ab 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/SimpleMessageStore.java +++ b/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> list() { + return new ArrayList>(this.map.values()); + } + public Message remove(Object key) { return (key != null) ? this.map.remove(key) : null; } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/message/SimpleMessageStoreTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/message/SimpleMessageStoreTests.java index ba3c57fcaf..a401eb2a6d 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/message/SimpleMessageStoreTests.java +++ b/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> 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> 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> list = store.list(); + assertEquals(2, list.size()); + assertEquals("bar", list.get(0).getPayload()); + assertEquals("baz", list.get(1).getPayload()); + } + }