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

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).

Browse Source
This commit is contained in:
Mark Fisher
2008-02-22 15:22:52 +00:00
parent 4226a980de
commit 094a131b1a
6 changed files with 198 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandlerAdapter.java
View File

@@ -39,21 +39,21 @@ import org.springframework.util.Assert;
*/
public abstract class AbstractMessageHandlerAdapter<T> 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<T> invoker;
private volatile SimpleMethodInvoker<T> 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<T> 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<T> 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<T> invoker);
}

4
spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java
View File

@@ -30,7 +30,7 @@ import org.springframework.integration.message.Message;
*/
public class MessageHandlerChain implements MessageHandler {
private List<MessageHandler> handlers = new CopyOnWriteArrayList<MessageHandler>();
private final List<MessageHandler> handlers = new CopyOnWriteArrayList<MessageHandler>();
/**
@@ -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) {

6
spring-integration-core/src/main/java/org/springframework/integration/router/RoutingBarrier.java
View File

@@ -34,15 +34,15 @@ import org.springframework.util.Assert;
*/
public class RoutingBarrier {
private List<Message<?>> messages = new CopyOnWriteArrayList<Message<?>>();
private final List<Message<?>> messages = new CopyOnWriteArrayList<Message<?>>();
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) {

61
spring-integration-core/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java
View File

@@ -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<T> extends AbstractMessageHandlerAdapter<T> implements ChannelRegistryAware {
public static final String CHANNEL_KEY = "channel";
private Map<String, ?> attributes;
private final Method method;
private Method method;
private final Map<String, ?> attributes;
private ChannelRegistry channelRegistry;
private volatile ChannelRegistry channelRegistry;
private long sendTimeout = -1;
private volatile long sendTimeout = -1;
public SplitterMessageHandlerAdapter(Object object, Method method, Map<String, ?> attributes) {
public SplitterMessageHandlerAdapter(T object, Method method, Map<String, ?> 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<T> 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()) {

14
spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagePublishingErrorHandler.java
View File

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

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

@@ -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<TestBean> adapter = new DefaultMessageHandlerAdapter<TestBean>();
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<TestBean> adapter = new DefaultMessageHandlerAdapter<TestBean>();
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<TestBean> adapter = new AbstractMessageHandlerAdapter<TestBean>() {
@Override
protected Object doHandle(Message message, SimpleMethodInvoker invoker) {
Object result = invoker.invokeMethod(message.getPayload());
Message<?> resultMessage = new GenericMessage<Object>(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<TestBean> adapter = new AbstractMessageHandlerAdapter<TestBean>() {
@Override
protected Object doHandle(Message message, SimpleMethodInvoker invoker) {
Object result = invoker.invokeMethod(message.getPayload());
Message<?> resultMessage = new GenericMessage<Object>(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<TestBean> adapter = new DefaultMessageHandlerAdapter<TestBean>();
adapter.setObject(new TestBean());
adapter.setMethodName("upperCase");
adapter.afterPropertiesSet();
MessageChannel testChannel = new SimpleChannel();
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
channelRegistry.registerChannel("testChannel", testChannel);
Map<String, String> attributes = new HashMap<String, String>();
attributes.put("channel", "testChannel");
SplitterMessageHandlerAdapter<TestBean> splitter = new SplitterMessageHandlerAdapter<TestBean>(
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(",");
}
}
}