Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Added ResponseCorrelator and RetrievalBlockingMessageStore (INT-143).

Browse Source
This commit is contained in:
Mark Fisher
2008-03-28 20:23:06 +00:00
parent 1fff0f0540
commit 51d080a058
6 changed files with 382 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
spring-integration-core/src/main/java/org/springframework/integration/handler/ResponseCorrelator.java Normal file
View File

@@ -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.
* <p>
* 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();
}
}

141
spring-integration-core/src/main/java/org/springframework/integration/message/RetrievalBlockingMessageStore.java Normal file
View File

@@ -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 <code>get</code> and
* <code>remove</code> methods block until a message is available.
* <p>
* Alternative methods that accept an explicit timeout value are also available.
*
* @author Mark Fisher
*/
public class RetrievalBlockingMessageStore extends SimpleMessageStore implements MessageStore {
private final ConcurrentMap<Object, List<SynchronousQueue<MessageHolder>>> listeners =
new ConcurrentHashMap<Object, List<SynchronousQueue<MessageHolder>>>();
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<SynchronousQueue<MessageHolder>> listenerList = null;
synchronized (this.listenerMonitor) {
listenerList = this.listeners.remove(key);
}
if (listenerList != null) {
for (int i = 0; i < listenerList.size(); i++) {
Queue<MessageHolder> 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<MessageHolder> queue = new SynchronousQueue<MessageHolder>();
synchronized (this.listenerMonitor) {
List<SynchronousQueue<MessageHolder>> listenerList = this.listeners.get(key);
if (listenerList == null) {
listenerList = new LinkedList<SynchronousQueue<MessageHolder>>();
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<SynchronousQueue<MessageHolder>> 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 <code>null</code> messages in the queue.
*/
private static class MessageHolder {
private final Message<?> message;
MessageHolder(Message<?> message) {
this.message = message;
}
Message<?> getMessage() {
return this.message;
}
}
}

7
spring-integration-core/src/main/java/org/springframework/integration/message/SimpleMessageStore.java
View File

@@ -28,16 +28,9 @@ import org.springframework.util.Assert;
*/
public class SimpleMessageStore implements MessageStore {
private static final int DEFAULT_CAPACITY = 1000;
private final Map<Object, Message<?>> map;
public SimpleMessageStore() {
this(DEFAULT_CAPACITY);
}
public SimpleMessageStore(int capacity) {
this.map = new BoundedHashMap<Object, Message<?>>(capacity);
}

82
spring-integration-core/src/test/java/org/springframework/integration/handler/ResponseCorrelatorTests.java Normal file
View File

@@ -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;
}
}

81
spring-integration-core/src/test/java/org/springframework/integration/message/RetrievalBlockingMessageStoreTests.java Normal file
View File

@@ -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));
}
});
}
}

6
spring-integration-core/src/test/java/org/springframework/integration/message/SimpleMessageStoreTests.java
View File

@@ -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);