diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/ResponseCorrelator.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/ResponseCorrelator.java new file mode 100644 index 0000000000..99383c88e9 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/ResponseCorrelator.java @@ -0,0 +1,75 @@ +/* + * 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.handler; + +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageHandlingException; +import org.springframework.integration.message.RetrievalBlockingMessageStore; +import org.springframework.util.Assert; + +/** + * A handler for receiving messages from a "reply channel". Any component that + * is expecting a response can poll by providing the correlation identifier. + * + * @author Mark Fisher + */ +public class ResponseCorrelator implements MessageHandler { + + private volatile long defaultTimeout = 5000; + + private final RetrievalBlockingMessageStore messageStore; + + + public ResponseCorrelator(int capacity) { + this.messageStore = new RetrievalBlockingMessageStore(capacity); + } + + + public void setDefaultTimeout(long defaultTimeout) { + Assert.isTrue(defaultTimeout >= 0, "'defaultTimeout' must not be negative"); + this.defaultTimeout = defaultTimeout; + } + + public Message handle(Message message) { + Object correlationId = this.getCorrelationId(message); + if (correlationId == null) { + throw new MessageHandlingException("unable to handle response, message has no correlationId: " + message); + } + this.messageStore.put(correlationId, message); + return null; + } + + public Message getResponse(Object correlationId) { + return this.getResponse(correlationId, this.defaultTimeout); + } + + public Message getResponse(Object correlationId, long timeout) { + Assert.notNull(correlationId, "'correlationId' must not be null"); + return this.messageStore.remove(correlationId, timeout); + } + + /** + * Retrieve the correlation identifier from the provided message. + *

+ * This method may be overridden by subclasses. The default implementation + * returns the 'correlationId' from the message header. + */ + protected Object getCorrelationId(final Message message) { + return message.getHeader().getCorrelationId(); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/RetrievalBlockingMessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/message/RetrievalBlockingMessageStore.java new file mode 100644 index 0000000000..a0b7099378 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/RetrievalBlockingMessageStore.java @@ -0,0 +1,141 @@ +/* + * 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.message; + +import java.util.LinkedList; +import java.util.List; +import java.util.Queue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.SynchronousQueue; +import java.util.concurrent.TimeUnit; + +/** + * A {@link MessageStore} implementation whose get and + * remove methods block until a message is available. + *

+ * Alternative methods that accept an explicit timeout value are also available. + * + * @author Mark Fisher + */ +public class RetrievalBlockingMessageStore extends SimpleMessageStore implements MessageStore { + + private final ConcurrentMap>> listeners = + new ConcurrentHashMap>>(); + + private final Object listenerMonitor = new Object(); + + + public RetrievalBlockingMessageStore(int capacity) { + super(capacity); + } + + + public Message put(Object key, Message message) { + Message previousMessage = super.put(key, message); + boolean sentReply = false; + List> listenerList = null; + synchronized (this.listenerMonitor) { + listenerList = this.listeners.remove(key); + } + if (listenerList != null) { + for (int i = 0; i < listenerList.size(); i++) { + Queue queue = listenerList.get(i); + if (!sentReply) { + sentReply = queue.offer(new MessageHolder(message)); + } + else { + queue.offer(new MessageHolder(null)); + } + } + } + return previousMessage; + } + + public Message get(Object key) { + return this.get(key, -1); + } + + public Message get(Object key, long timeout) { + Message message = super.get(key); + return (message != null) ? message : waitForMessage(key, timeout, false); + } + + public Message remove(Object key) { + return this.remove(key, -1); + } + + public Message remove(Object key, long timeout) { + Message message = super.remove(key); + return (message != null) ? message : waitForMessage(key, timeout, true); + } + + private Message waitForMessage(Object key, long timeout, boolean shouldRemove) { + Message message = null; + SynchronousQueue queue = new SynchronousQueue(); + synchronized (this.listenerMonitor) { + List> listenerList = this.listeners.get(key); + if (listenerList == null) { + listenerList = new LinkedList>(); + this.listeners.put(key, listenerList); + } + listenerList.add(queue); + } + try { + MessageHolder holder = (timeout < 0) ? queue.take() : queue.poll(timeout, TimeUnit.MILLISECONDS); + if (holder != null) { + message = holder.getMessage(); + if (message != null && shouldRemove) { + super.remove(key); + } + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + finally { + synchronized (this.listenerMonitor) { + List> listenerList = this.listeners.get(key); + if (listenerList != null) { + listenerList.remove(queue); + if (listenerList.size() == 0) { + this.listeners.remove(key); + } + } + } + } + return message; + } + + + /** + * A wrapper class to enable null messages in the queue. + */ + private static class MessageHolder { + + private final Message message; + + MessageHolder(Message message) { + this.message = message; + } + + Message getMessage() { + return this.message; + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/SimpleMessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/message/SimpleMessageStore.java index 44a9dd0fd4..fbe10830d3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/message/SimpleMessageStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/SimpleMessageStore.java @@ -28,16 +28,9 @@ import org.springframework.util.Assert; */ public class SimpleMessageStore implements MessageStore { - private static final int DEFAULT_CAPACITY = 1000; - - private final Map> map; - public SimpleMessageStore() { - this(DEFAULT_CAPACITY); - } - public SimpleMessageStore(int capacity) { this.map = new BoundedHashMap>(capacity); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/ResponseCorrelatorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/ResponseCorrelatorTests.java new file mode 100644 index 0000000000..8bffc7bc27 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/ResponseCorrelatorTests.java @@ -0,0 +1,82 @@ +/* + * 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.handler; + +import static org.junit.Assert.assertEquals; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executor; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.Test; + +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + */ +public class ResponseCorrelatorTests { + + @Test + public void testReceiversPrecedeResponse() throws InterruptedException { + final ResponseCorrelator correlator = new ResponseCorrelator(10); + final AtomicInteger responseCounter = new AtomicInteger(); + CountDownLatch latch = startReceivers(correlator, responseCounter, 5, 500); + Message message = new StringMessage("test"); + message.getHeader().setCorrelationId("123"); + correlator.handle(message); + latch.await(1000, TimeUnit.MILLISECONDS); + assertEquals(0, latch.getCount()); + assertEquals(1, responseCounter.get()); + } + + @Test + public void testResponsePrecedeReceivers() throws InterruptedException { + final ResponseCorrelator correlator = new ResponseCorrelator(10); + Message message = new StringMessage("test"); + message.getHeader().setCorrelationId("123"); + correlator.handle(message); + final AtomicInteger responseCounter = new AtomicInteger(); + CountDownLatch latch = startReceivers(correlator, responseCounter, 5, 50); + latch.await(1000, TimeUnit.MILLISECONDS); + assertEquals(0, latch.getCount()); + assertEquals(1, responseCounter.get()); + } + + + private static CountDownLatch startReceivers(final ResponseCorrelator correlator, + final AtomicInteger responseCounter, int numReceivers, final long timeout) { + final CountDownLatch latch = new CountDownLatch(numReceivers); + Executor executor = Executors.newFixedThreadPool(numReceivers); + for (int i = 0; i < numReceivers; i++) { + executor.execute(new Runnable() { + public void run() { + Message response = correlator.getResponse("123", timeout); + if (response != null) { + responseCounter.incrementAndGet(); + } + latch.countDown(); + } + }); + } + return latch; + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/message/RetrievalBlockingMessageStoreTests.java b/spring-integration-core/src/test/java/org/springframework/integration/message/RetrievalBlockingMessageStoreTests.java new file mode 100644 index 0000000000..07f4d4bc04 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/message/RetrievalBlockingMessageStoreTests.java @@ -0,0 +1,81 @@ +/* + * 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.message; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; + +import java.util.concurrent.Executors; + +import org.junit.Test; + +/** + * @author Mark Fisher + */ +public class RetrievalBlockingMessageStoreTests { + + @Test + public void testGetWithElapsedTimeout() { + final RetrievalBlockingMessageStore store = new RetrievalBlockingMessageStore(10); + publishWithDelay(store, "foo", "bar", 100); + Message message = store.get("foo", 5); + assertNull(message); + } + + @Test + public void testGetWithinTimeout() { + final RetrievalBlockingMessageStore store = new RetrievalBlockingMessageStore(10); + publishWithDelay(store, "foo", "bar", 50); + Message message = store.get("foo", 500); + assertNotNull(message); + assertEquals("bar", message.getPayload()); + assertNotNull(store.get("foo", 0)); + } + + @Test + public void testRemoveWithElapsedTimeout() { + final RetrievalBlockingMessageStore store = new RetrievalBlockingMessageStore(10); + publishWithDelay(store, "foo", "bar", 100); + Message message = store.remove("foo", 5); + assertNull(message); + } + + @Test + public void testRemoveWithinTimeout() { + final RetrievalBlockingMessageStore store = new RetrievalBlockingMessageStore(10); + publishWithDelay(store, "foo", "bar", 50); + Message message = store.remove("foo", 500); + assertNotNull(message); + assertEquals("bar", message.getPayload()); + assertNull(store.get("foo", 0)); + } + + + private static void publishWithDelay(final MessageStore store, final String key, final String value, final long delay) { + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + Thread.sleep(delay); + } + catch (InterruptedException e) {} + store.put(key, new StringMessage(value)); + } + }); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/message/SimpleMessageStoreTests.java b/spring-integration-core/src/test/java/org/springframework/integration/message/SimpleMessageStoreTests.java index 9f415ae12c..ba3c57fcaf 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/message/SimpleMessageStoreTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/message/SimpleMessageStoreTests.java @@ -29,7 +29,7 @@ public class SimpleMessageStoreTests { @Test public void testPut() { - SimpleMessageStore store = new SimpleMessageStore(); + SimpleMessageStore store = new SimpleMessageStore(5); Message message1 = new StringMessage("message-1"); Message previous = store.put(1, message1); assertNull(previous); @@ -39,7 +39,7 @@ public class SimpleMessageStoreTests { @Test public void testReplace() { - SimpleMessageStore store = new SimpleMessageStore(); + SimpleMessageStore store = new SimpleMessageStore(5); Message messageA = new StringMessage("message-a"); Message messageB = new StringMessage("message-b"); store.put(1, messageA); @@ -50,7 +50,7 @@ public class SimpleMessageStoreTests { @Test public void testRemove() { - SimpleMessageStore store = new SimpleMessageStore(); + SimpleMessageStore store = new SimpleMessageStore(5); Message message = new StringMessage("message"); assertNull(store.remove(1)); store.put(1, message);