From d38a7259d3cdbdae1279166162dd36fe06c7c1a4 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 15 May 2008 20:08:04 +0000 Subject: [PATCH] ResponseCorrelator is now ReplyMessageCorrelator. --- .../channel/RequestReplyTemplate.java | 28 ++++++++-------- ...lator.java => ReplyMessageCorrelator.java} | 12 +++---- ....java => ReplyMessageCorrelatorTests.java} | 32 +++++++++---------- 3 files changed, 36 insertions(+), 36 deletions(-) rename spring-integration-core/src/main/java/org/springframework/integration/handler/{ResponseCorrelator.java => ReplyMessageCorrelator.java} (85%) rename spring-integration-core/src/test/java/org/springframework/integration/handler/{ResponseCorrelatorTests.java => ReplyMessageCorrelatorTests.java} (65%) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/RequestReplyTemplate.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/RequestReplyTemplate.java index 1d14d7f9ca..425b97aff6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/RequestReplyTemplate.java +++ b/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; } } 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/ReplyMessageCorrelator.java similarity index 85% rename from spring-integration-core/src/main/java/org/springframework/integration/handler/ResponseCorrelator.java rename to spring-integration-core/src/main/java/org/springframework/integration/handler/ReplyMessageCorrelator.java index 9f64ec415b..be62236058 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/ResponseCorrelator.java +++ b/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); } 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/ReplyMessageCorrelatorTests.java similarity index 65% rename from spring-integration-core/src/test/java/org/springframework/integration/handler/ResponseCorrelatorTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/handler/ReplyMessageCorrelatorTests.java index 8bffc7bc27..f850c174b0 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/ResponseCorrelatorTests.java +++ b/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(); }