ResponseCorrelator is now ReplyMessageCorrelator.

spring-integration-core/src/main/java/org/springframework/integration/channel/RequestReplyTemplate.java

@@ -24,7 +24,7 @@ import org.springframework.integration.bus.MessageBusAware;
 import org.springframework.integration.endpoint.EndpointRegistry;
 import org.springframework.integration.endpoint.HandlerEndpoint;
 import org.springframework.integration.handler.ReplyHandler;
-import org.springframework.integration.handler.ResponseCorrelator;
+import org.springframework.integration.handler.ReplyMessageCorrelator;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessagingException;
 import org.springframework.integration.message.selector.MessageSelector;
@@ -46,11 +46,11 @@ public class RequestReplyTemplate implements MessageBusAware {
 
 	private volatile long replyTimeout = -1;
 
-	private ResponseCorrelator responseCorrelator;
+	private ReplyMessageCorrelator replyMessageCorrelator;
 
 	private EndpointRegistry endpointRegistry;
 
-	private final Object responseCorrelatorMonitor = new Object();
+	private final Object replyMessageCorrelatorMonitor = new Object();
 
 
 	/**
@@ -170,21 +170,21 @@ public class RequestReplyTemplate implements MessageBusAware {
 			throw new MessagingException("No request channel available. Cannot send request message.");
 		}
 		if (this.replyChannel != null) {
-			return this.sendAndReceiveWithResponseCorrelator(message);
+			return this.sendAndReceiveWithReplyMessageCorrelator(message);
 		}
 		else {
 			return this.sendAndReceiveWithTemporaryChannel(message);
 		}
 	}
 
-	private Message<?> sendAndReceiveWithResponseCorrelator(Message<?> message) {
-		if (this.responseCorrelator == null) {
-			this.registerResponseCorrelator();
+	private Message<?> sendAndReceiveWithReplyMessageCorrelator(Message<?> message) {
+		if (this.replyMessageCorrelator == null) {
+			this.registerReplyMessageCorrelator();
 		}
 		message.getHeader().setReturnAddress(this.replyChannel);
 		this.send(message);
-		return (this.replyTimeout >= 0) ? this.responseCorrelator.getResponse(message.getId(), this.replyTimeout) :
-				this.responseCorrelator.getResponse(message.getId());
+		return (this.replyTimeout >= 0) ? this.replyMessageCorrelator.getReply(message.getId(), this.replyTimeout) :
+				this.replyMessageCorrelator.getReply(message.getId());
 	}
 
 	private Message<?> sendAndReceiveWithTemporaryChannel(Message<?> message) {
@@ -198,19 +198,19 @@ public class RequestReplyTemplate implements MessageBusAware {
 		return (this.replyTimeout >= 0) ? channel.receive(this.replyTimeout) : channel.receive();
 	}
 
-	private void registerResponseCorrelator() {
-		synchronized (this.responseCorrelatorMonitor) {
-			if (this.responseCorrelator != null) {
+	private void registerReplyMessageCorrelator() {
+		synchronized (this.replyMessageCorrelatorMonitor) {
+			if (this.replyMessageCorrelator != null) {
 				return;
 			}
 			if (this.endpointRegistry == null) {
 				throw new ConfigurationException("No EndpointRegistry available. Cannot register ResponseCorrelator.");
 			}
-			ResponseCorrelator correlator = new ResponseCorrelator(10);
+			ReplyMessageCorrelator correlator = new ReplyMessageCorrelator(10);
 			HandlerEndpoint endpoint = new HandlerEndpoint(correlator);
 			endpoint.setSubscription(new Subscription(this.replyChannel));
 			this.endpointRegistry.registerEndpoint("internal.correlator." + this, endpoint);
-			this.responseCorrelator = correlator;
+			this.replyMessageCorrelator = correlator;
 		}
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/handler/ReplyMessageCorrelator.java

@@ -23,18 +23,18 @@ 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.
+ * is expecting a reply message can poll by providing the correlation identifier.
  * 
  * @author Mark Fisher
  */
-public class ResponseCorrelator implements MessageHandler {
+public class ReplyMessageCorrelator implements MessageHandler {
 
 	private volatile long defaultTimeout = 5000;
 
 	private final RetrievalBlockingMessageStore messageStore;
 
 
-	public ResponseCorrelator(int capacity) {
+	public ReplyMessageCorrelator(int capacity) {
 		this.messageStore = new RetrievalBlockingMessageStore(capacity);
 	}
 
@@ -54,11 +54,11 @@ public class ResponseCorrelator implements MessageHandler {
 		return null;
 	}
 
-	public Message<?> getResponse(Object correlationId) {
-		return this.getResponse(correlationId, this.defaultTimeout);
+	public Message<?> getReply(Object correlationId) {
+		return this.getReply(correlationId, this.defaultTimeout);
 	}
 
-	public Message<?> getResponse(Object correlationId, long timeout) {
+	public Message<?> getReply(Object correlationId, long timeout) {
 		Assert.notNull(correlationId, "'correlationId' must not be null");
 		return this.messageStore.remove(correlationId, timeout);
 	}

spring-integration-core/src/test/java/org/springframework/integration/handler/ReplyMessageCorrelatorTests.java

@@ -32,45 +32,45 @@ import org.springframework.integration.message.StringMessage;
 /**
  * @author Mark Fisher
  */
-public class ResponseCorrelatorTests {
+public class ReplyMessageCorrelatorTests {
 
 	@Test
-	public void testReceiversPrecedeResponse() throws InterruptedException {
-		final ResponseCorrelator correlator = new ResponseCorrelator(10);
-		final AtomicInteger responseCounter = new AtomicInteger();
-		CountDownLatch latch = startReceivers(correlator, responseCounter, 5, 500);
+	public void testReceiversPrecedeReply() throws InterruptedException {
+		final ReplyMessageCorrelator correlator = new ReplyMessageCorrelator(10);
+		final AtomicInteger replyCounter = new AtomicInteger();
+		CountDownLatch latch = startReceivers(correlator, replyCounter, 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());
+		assertEquals(1, replyCounter.get());
 	}
 
 	@Test
-	public void testResponsePrecedeReceivers() throws InterruptedException {
-		final ResponseCorrelator correlator = new ResponseCorrelator(10);
+	public void testReplyPrecedeReceivers() throws InterruptedException {
+		final ReplyMessageCorrelator correlator = new ReplyMessageCorrelator(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);
+		final AtomicInteger replyCounter = new AtomicInteger();
+		CountDownLatch latch = startReceivers(correlator, replyCounter, 5, 50);
 		latch.await(1000, TimeUnit.MILLISECONDS);
 		assertEquals(0, latch.getCount());
-		assertEquals(1, responseCounter.get());
+		assertEquals(1, replyCounter.get());
 	}
 
 
-	private static CountDownLatch startReceivers(final ResponseCorrelator correlator,
-			final AtomicInteger responseCounter, int numReceivers, final long timeout) {
+	private static CountDownLatch startReceivers(final ReplyMessageCorrelator correlator,
+			final AtomicInteger replyCounter, 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();
+					Message<?> reply = correlator.getReply("123", timeout);
+					if (reply != null) {
+						replyCounter.incrementAndGet();
 					}
 					latch.countDown();
 				}