From 094a131b1a3d4ba426a651ddebb6f8a193010eba Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 22 Feb 2008 15:22:52 +0000 Subject: [PATCH] The 'correlationId' is now being handled by the AbstractMessageHandlerAdapter (see INT-127). SplitterMessageHandlerAdapter now properly sets the 'sequenceNumber' and 'sequenceSize' properties on the reply's MessageHeader (INT-128). --- .../AbstractMessageHandlerAdapter.java | 27 ++-- .../handler/MessageHandlerChain.java | 4 +- .../integration/router/RoutingBarrier.java | 6 +- .../router/SplitterMessageHandlerAdapter.java | 61 ++++---- .../MessagePublishingErrorHandler.java | 14 +- .../handler/CorrelationIdTests.java | 140 ++++++++++++++++++ 6 files changed, 198 insertions(+), 54 deletions(-) create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandlerAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandlerAdapter.java index 9761f80e28..33475b8b44 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandlerAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandlerAdapter.java @@ -39,21 +39,21 @@ import org.springframework.util.Assert; */ public abstract class AbstractMessageHandlerAdapter implements MessageHandler, Ordered, InitializingBean { - protected Log logger = LogFactory.getLog(this.getClass()); + protected final Log logger = LogFactory.getLog(this.getClass()); - private T object; + private volatile T object; - private String methodName; + private volatile String methodName; - private MessageMapper mapper = new SimplePayloadMessageMapper(); + private volatile MessageMapper mapper = new SimplePayloadMessageMapper(); - private SimpleMethodInvoker invoker; + private volatile SimpleMethodInvoker invoker; - private int order = Integer.MAX_VALUE; + private volatile int order = Integer.MAX_VALUE; private volatile boolean initialized; - private Object lifecycleMonitor = new Object(); + private final Object lifecycleMonitor = new Object(); public void setObject(T object) { @@ -108,10 +108,15 @@ public abstract class AbstractMessageHandlerAdapter implements MessageHandler public final Message handle(Message message) { Object result = this.doHandle(message, invoker); if (result != null) { - if (result instanceof Message) { - return (Message) result; + Message reply = (result instanceof Message) ? (Message) result : + this.mapper.toMessage(result); + Object correlationId = reply.getHeader().getCorrelationId(); + if (correlationId == null) { + Object orginalCorrelationId = message.getHeader().getCorrelationId(); + reply.getHeader().setCorrelationId((orginalCorrelationId != null) ? + orginalCorrelationId : message.getId()); } - return this.mapper.toMessage(result); + return reply; } return null; } @@ -127,6 +132,6 @@ public abstract class AbstractMessageHandlerAdapter implements MessageHandler * the provided target object and method. May return an object of type * {@link Message}, else rely on the message mapper to convert. */ - protected abstract Object doHandle(Message message, SimpleMethodInvoker invoker); + protected abstract Object doHandle(Message message, SimpleMethodInvoker invoker); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java index 5d5e573a8d..82cdec27f1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java @@ -30,7 +30,7 @@ import org.springframework.integration.message.Message; */ public class MessageHandlerChain implements MessageHandler { - private List handlers = new CopyOnWriteArrayList(); + private final List handlers = new CopyOnWriteArrayList(); /** @@ -52,7 +52,7 @@ public class MessageHandlerChain implements MessageHandler { this.handlers.addAll(handlers); } - public Message handle(Message message) { + public final Message handle(Message message) { for (MessageHandler next : handlers) { message = next.handle(message); if (message == null) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/RoutingBarrier.java b/spring-integration-core/src/main/java/org/springframework/integration/router/RoutingBarrier.java index 2b8f7ef9ac..650f9e107b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/RoutingBarrier.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/RoutingBarrier.java @@ -34,15 +34,15 @@ import org.springframework.util.Assert; */ public class RoutingBarrier { - private List> messages = new CopyOnWriteArrayList>(); + private final List> messages = new CopyOnWriteArrayList>(); private final RoutingBarrierCompletionStrategy completionStrategy; private volatile boolean complete = false; - private ReentrantLock lock = new ReentrantLock(); + private final ReentrantLock lock = new ReentrantLock(); - private Condition condition = lock.newCondition(); + private final Condition condition = lock.newCondition(); public RoutingBarrier(RoutingBarrierCompletionStrategy completionStrategy) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java index 71562af441..96e9d30c7b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java @@ -35,22 +35,23 @@ import org.springframework.util.Assert; * MessageHandler adapter for methods annotated with {@link Splitter @Splitter}. * * @author Mark Fisher + * @author Marius Bogoevici */ -public class SplitterMessageHandlerAdapter extends AbstractMessageHandlerAdapter implements ChannelRegistryAware { +public class SplitterMessageHandlerAdapter extends AbstractMessageHandlerAdapter implements ChannelRegistryAware { public static final String CHANNEL_KEY = "channel"; - private Map attributes; + private final Method method; - private Method method; + private final Map attributes; - private ChannelRegistry channelRegistry; + private volatile ChannelRegistry channelRegistry; - private long sendTimeout = -1; + private volatile long sendTimeout = -1; - public SplitterMessageHandlerAdapter(Object object, Method method, Map attributes) { + public SplitterMessageHandlerAdapter(T object, Method method, Map attributes) { Assert.notNull(object, "'object' must not be null"); Assert.notNull(method, "'method' must not be null"); Assert.isTrue(attributes != null && attributes.get(CHANNEL_KEY) != null, @@ -70,10 +71,10 @@ public class SplitterMessageHandlerAdapter extends AbstractMessageHandlerAdapter } @Override - protected Object doHandle(Message message, SimpleMethodInvoker invoker) { + protected final Object doHandle(Message message, SimpleMethodInvoker invoker) { if (method.getParameterTypes().length != 1) { throw new MessagingConfigurationException( - "method must accept exactly one parameter"); + "Splitter method must accept exactly one parameter"); } String channelName = (String) attributes.get(CHANNEL_KEY); Object retval = null; @@ -98,33 +99,20 @@ public class SplitterMessageHandlerAdapter extends AbstractMessageHandlerAdapter } if (retval instanceof Collection) { Collection items = (Collection) retval; - int counter = 0; + int sequenceNumber = 0; + int sequenceSize = items.size(); for (Object item : items) { - if (item instanceof Message) { - Message splitMessage = (Message) item; - splitMessage.getHeader().setCorrelationId(message.getId()); - this.sendMessage(splitMessage, channelName); - } - else { - Message splitMessage = new GenericMessage(message.getId() + "#" + (counter++), item); - splitMessage.getHeader().setCorrelationId(message.getId()); - this.sendMessage(splitMessage, channelName); - } + Message splitMessage = prepareMessage(item, message.getId(), ++sequenceNumber, sequenceSize); + this.sendMessage(splitMessage, channelName); } } else if (retval.getClass().isArray()) { - int counter = 0; - for (Object item : (Object[]) retval) { - if (item instanceof Message) { - Message splitMessage = (Message) item; - splitMessage.getHeader().setCorrelationId(message.getId()); - this.sendMessage(splitMessage, channelName); - } - else { - Message splitMessage = new GenericMessage(message.getId() + "#" + (counter++), item); - splitMessage.getHeader().setCorrelationId(message.getId()); - this.sendMessage(splitMessage, channelName); - } + Object[] array = (Object[]) retval; + int sequenceNumber = 0; + int sequenceSize = array.length; + for (Object item : array) { + Message splitMessage = prepareMessage(item, message.getId(), ++sequenceNumber, sequenceSize); + this.sendMessage(splitMessage, channelName); } } else { @@ -134,7 +122,18 @@ public class SplitterMessageHandlerAdapter extends AbstractMessageHandlerAdapter return null; } + private Message prepareMessage(Object item, Object correlationId, int sequenceNumber, int sequenceSize) { + Message message = (item instanceof Message) ? (Message) item : new GenericMessage(item); + message.getHeader().setCorrelationId(correlationId); + message.getHeader().setSequenceNumber(sequenceNumber); + message.getHeader().setSequenceSize(sequenceSize); + return message; + } + private boolean sendMessage(Message message, String channelName) { + if (this.channelRegistry == null) { + throw new IllegalStateException(this.getClass().getSimpleName() + " requires a ChannelRegistry reference."); + } MessageChannel channel = this.channelRegistry.lookupChannel(channelName); if (channel == null) { if (logger.isWarnEnabled()) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagePublishingErrorHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagePublishingErrorHandler.java index c76387e102..8015d6c144 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagePublishingErrorHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagePublishingErrorHandler.java @@ -31,11 +31,11 @@ import org.springframework.integration.util.ErrorHandler; */ public class MessagePublishingErrorHandler implements ErrorHandler { - private Log logger = LogFactory.getLog(this.getClass()); + private final Log logger = LogFactory.getLog(this.getClass()); - private MessageChannel errorChannel; + private volatile MessageChannel errorChannel; - private long sendTimeout = 1000; + private final long sendTimeout = 1000; public MessagePublishingErrorHandler() { @@ -50,7 +50,10 @@ public class MessagePublishingErrorHandler implements ErrorHandler { this.errorChannel = errorChannel; } - public void handle(Throwable t) { + public final void handle(Throwable t) { + if (logger.isWarnEnabled()) { + logger.warn("failure occurred in messaging task", t); + } if (this.errorChannel != null) { try { this.errorChannel.send(new ErrorMessage(t), this.sendTimeout); @@ -58,9 +61,6 @@ public class MessagePublishingErrorHandler implements ErrorHandler { catch (Throwable ignore) { // message will be logged only } } - if (logger.isWarnEnabled()) { - logger.warn("failure occurred in messaging task", t); - } } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java new file mode 100644 index 0000000000..57c139955d --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java @@ -0,0 +1,140 @@ +/* + * Copyright 2002-2007 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.HashMap; +import java.util.Map; + +import org.junit.Test; + +import org.springframework.integration.channel.ChannelRegistry; +import org.springframework.integration.channel.DefaultChannelRegistry; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; +import org.springframework.integration.router.SplitterMessageHandlerAdapter; +import org.springframework.integration.util.SimpleMethodInvoker; + +/** + * @author Mark Fisher + */ +public class CorrelationIdTests { + + @Test + public void testCorrelationIdPassedIfAvailable() { + Object correlationId = "123-ABC"; + Message message = new StringMessage("test"); + message.getHeader().setCorrelationId(correlationId); + DefaultMessageHandlerAdapter adapter = new DefaultMessageHandlerAdapter(); + adapter.setObject(new TestBean()); + adapter.setMethodName("upperCase"); + adapter.afterPropertiesSet(); + Message reply = adapter.handle(message); + assertEquals(correlationId, reply.getHeader().getCorrelationId()); + } + + @Test + public void testCorrelationIdCopiedFromMessageIdByDefault() { + Message message = new StringMessage("test"); + DefaultMessageHandlerAdapter adapter = new DefaultMessageHandlerAdapter(); + adapter.setObject(new TestBean()); + adapter.setMethodName("upperCase"); + adapter.afterPropertiesSet(); + Message reply = adapter.handle(message); + assertEquals(message.getId(), reply.getHeader().getCorrelationId()); + } + + @Test + public void testCorrelationNotPassedIfAlreadySetByHandler() throws Exception { + Object correlationId = "123-ABC"; + Message message = new StringMessage("test"); + message.getHeader().setCorrelationId(correlationId); + AbstractMessageHandlerAdapter adapter = new AbstractMessageHandlerAdapter() { + @Override + protected Object doHandle(Message message, SimpleMethodInvoker invoker) { + Object result = invoker.invokeMethod(message.getPayload()); + Message resultMessage = new GenericMessage(result); + resultMessage.getHeader().setCorrelationId("456-XYZ"); + return resultMessage; + } + }; + adapter.setObject(new TestBean()); + adapter.setMethodName("upperCase"); + adapter.afterPropertiesSet(); + Message reply = adapter.handle(message); + assertEquals("456-XYZ", reply.getHeader().getCorrelationId()); + } + + @Test + public void testCorrelationNotCopiedIfAlreadySetByHandler() throws Exception { + Message message = new StringMessage("test"); + AbstractMessageHandlerAdapter adapter = new AbstractMessageHandlerAdapter() { + @Override + protected Object doHandle(Message message, SimpleMethodInvoker invoker) { + Object result = invoker.invokeMethod(message.getPayload()); + Message resultMessage = new GenericMessage(result); + resultMessage.getHeader().setCorrelationId("456-XYZ"); + return resultMessage; + } + }; + adapter.setObject(new TestBean()); + adapter.setMethodName("upperCase"); + adapter.afterPropertiesSet(); + Message reply = adapter.handle(message); + assertEquals("456-XYZ", reply.getHeader().getCorrelationId()); + } + + @Test + public void testCorrelationIdWithSplitter() throws Exception { + Message message = new StringMessage("test1,test2"); + DefaultMessageHandlerAdapter adapter = new DefaultMessageHandlerAdapter(); + adapter.setObject(new TestBean()); + adapter.setMethodName("upperCase"); + adapter.afterPropertiesSet(); + MessageChannel testChannel = new SimpleChannel(); + ChannelRegistry channelRegistry = new DefaultChannelRegistry(); + channelRegistry.registerChannel("testChannel", testChannel); + Map attributes = new HashMap(); + attributes.put("channel", "testChannel"); + SplitterMessageHandlerAdapter splitter = new SplitterMessageHandlerAdapter( + new TestBean(), TestBean.class.getMethod("split", String.class), attributes); + splitter.setChannelRegistry(channelRegistry); + splitter.afterPropertiesSet(); + splitter.handle(message); + Message reply1 = testChannel.receive(100); + Message reply2 = testChannel.receive(100); + assertEquals(message.getId(), reply1.getHeader().getCorrelationId()); + assertEquals(message.getId(), reply2.getHeader().getCorrelationId()); + } + + + private static class TestBean { + + public String upperCase(String input) { + return input.toUpperCase(); + } + + public String[] split(String input) { + return input.split(","); + } + } + +}