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

INT-901 removed ReplyMessageHolder and refactored AbstractReplyProducingMessageHandler accordingly

Browse Source
This commit is contained in:
Mark Fisher
2009-12-02 02:37:51 +00:00
parent 40f80d96cb
commit 4a13e38fde
17 changed files with 118 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
org.springframework.integration/src/main/java/org/springframework/integration/filter/MessageFilter.java
View File

@@ -19,7 +19,6 @@ package org.springframework.integration.filter;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.ReplyMessageHolder;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageRejectedException;
import org.springframework.integration.selector.MessageSelector;
@@ -83,22 +82,21 @@ public class MessageFilter extends AbstractReplyProducingMessageHandler {
}
@Override
protected void handleRequestMessage(Message<?> message, ReplyMessageHolder replyHolder) {
protected Object handleRequestMessage(Message<?> message) {
if (this.selector.accept(message)) {
replyHolder.set(message);
return message;
}
else {
if (this.discardChannel != null) {
boolean discarded = this.sendReplyMessage(message, this.discardChannel);
if (!discarded) {
throw new MessageDeliveryException(message,
"failed to send rejected Message to the discard channel");
}
}
if (this.throwExceptionOnRejection) {
throw new MessageRejectedException(message);
if (this.discardChannel != null) {
boolean discarded = this.sendReplyMessage(message, this.discardChannel);
if (!discarded) {
throw new MessageDeliveryException(message,
"failed to send rejected Message to the discard channel");
}
}
if (this.throwExceptionOnRejection) {
throw new MessageRejectedException(message);
}
return null;
}
}

43
org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2008 the original author or authors.
* Copyright 2002-2009 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.
@@ -96,10 +96,10 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
* {@inheritDoc}
*/
@Override
@SuppressWarnings("unchecked")
protected final void handleMessageInternal(Message<?> message) {
ReplyMessageHolder replyMessageHolder = new ReplyMessageHolder();
this.handleRequestMessage(message, replyMessageHolder);
if (replyMessageHolder.isEmpty()) {
Object result = this.handleRequestMessage(message);
if (result == null) {
if (this.requiresReply) {
throw new MessageHandlingException(message, "handler '" + this
+ "' requires a reply, but no reply was received");
@@ -111,9 +111,11 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
}
MessageChannel replyChannel = resolveReplyChannel(message, this.outputChannel, this.channelResolver);
MessageHeaders requestHeaders = message.getHeaders();
for (MessageBuilder<?> builder : replyMessageHolder.builders()) {
builder.copyHeadersIfAbsent(requestHeaders);
Message<?> replyMessage = builder.build();
if (result instanceof Iterable) {
this.sendReplyMessages((Iterable) result, requestHeaders, replyChannel);
}
else {
Message<?> replyMessage = this.createReplyMessage(result, requestHeaders);
if (!this.sendReplyMessage(replyMessage, replyChannel)) {
throw new MessageDeliveryException(replyMessage,
"failed to send reply Message to channel '" + replyChannel + "'");
@@ -121,7 +123,24 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
}
}
protected abstract void handleRequestMessage(Message<?> requestMessage, ReplyMessageHolder replyMessageHolder);
private void sendReplyMessages(Iterable<?> replies, MessageHeaders requestHeaders, MessageChannel replyChannel) {
for (Object reply : replies) {
Message<?> replyMessage = this.createReplyMessage(reply, requestHeaders);
if (!this.sendReplyMessage(replyMessage, replyChannel)) {
throw new MessageDeliveryException(replyMessage,
"failed to send reply Message to channel '" + replyChannel + "'");
}
}
}
@SuppressWarnings("unchecked")
private Message<?> createReplyMessage(Object reply, MessageHeaders requestHeaders) {
MessageBuilder<?> builder = (reply instanceof MessageBuilder) ? (MessageBuilder) reply
: (reply instanceof Message) ? MessageBuilder.fromMessage((Message) reply)
: MessageBuilder.withPayload(reply);
builder.copyHeadersIfAbsent(requestHeaders);
return builder.build();
}
protected boolean sendReplyMessage(Message<?> replyMessage, MessageChannel replyChannel) {
if (logger.isDebugEnabled()) {
@@ -130,4 +149,12 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
return this.channelTemplate.send(replyMessage, replyChannel);
}
/**
* Subclasses must implement this method to handle the request Message. The return
* value may be a Message, a MessageBuilder, or any plain Object. The base class
* will handle the final creation of a reply Message from any of those starting
* points. If the return value is null, the Message flow will end here.
*/
protected abstract Object handleRequestMessage(Message<?> requestMessage);
}

4
org.springframework.integration/src/main/java/org/springframework/integration/handler/BridgeHandler.java
View File

@@ -36,11 +36,11 @@ import org.springframework.util.Assert;
public class BridgeHandler extends AbstractReplyProducingMessageHandler {
@Override
protected void handleRequestMessage(Message<?> requestMessage, ReplyMessageHolder replyMessageHolder) {
protected Object handleRequestMessage(Message<?> requestMessage) {
if (requestMessage.getHeaders().getReplyChannel() == null) {
this.verifyOutputChannel();
}
replyMessageHolder.set(requestMessage);
return requestMessage;
}
private void verifyOutputChannel() {

70
org.springframework.integration/src/main/java/org/springframework/integration/handler/ReplyMessageHolder.java
View File

@@ -1,70 +0,0 @@
/*
* 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 java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageBuilder;
/**
* @author Mark Fisher
*/
public class ReplyMessageHolder {
private final List<MessageBuilder<?>> builders = new ArrayList<MessageBuilder<?>>();
public MessageBuilder<?> set(Object messageOrPayload) {
return this.createAndAddBuilder(messageOrPayload, true);
}
public MessageBuilder<?> add(Object messageOrPayload) {
return this.createAndAddBuilder(messageOrPayload, false);
}
public boolean isEmpty() {
return this.builders.isEmpty();
}
public List<MessageBuilder<?>> builders() {
return Collections.unmodifiableList(this.builders);
}
private MessageBuilder<?> createAndAddBuilder(Object messageOrPayload, boolean clearExistingValues) {
MessageBuilder<?> builder = null;
if (messageOrPayload instanceof MessageBuilder) {
builder = (MessageBuilder<?>) messageOrPayload;
}
else if (messageOrPayload instanceof Message) {
builder = MessageBuilder.fromMessage((Message<?>) messageOrPayload);
}
else {
builder = MessageBuilder.withPayload(messageOrPayload);
}
synchronized (this.builders) {
if (clearExistingValues) {
this.builders.clear();
}
this.builders.add(builder);
}
return builder;
}
}

7
org.springframework.integration/src/main/java/org/springframework/integration/handler/ServiceActivatingHandler.java
View File

@@ -44,12 +44,9 @@ public class ServiceActivatingHandler extends AbstractReplyProducingMessageHandl
@Override
protected void handleRequestMessage(Message<?> message, ReplyMessageHolder replyHolder) {
protected Object handleRequestMessage(Message<?> message) {
try {
Object result = this.processor.processMessage(message);
if (result != null) {
replyHolder.set(result);
}
return this.processor.processMessage(message);
}
catch (Exception e) {
if (e instanceof RuntimeException) {

26
org.springframework.integration/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java
View File

@@ -16,13 +16,15 @@
package org.springframework.integration.splitter;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.UUID;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.ReplyMessageHolder;
import org.springframework.integration.message.MessageBuilder;
/**
* Base class for Message-splitting handlers.
@@ -32,19 +34,20 @@ import org.springframework.integration.handler.ReplyMessageHolder;
public abstract class AbstractMessageSplitter extends AbstractReplyProducingMessageHandler {
@Override
protected final void handleRequestMessage(Message<?> message, ReplyMessageHolder replyHolder) {
protected final Object handleRequestMessage(Message<?> message) {
Object result = this.splitMessage(message);
if (result == null) {
return;
return null;
}
Object correlationId = (message.getHeaders().getCorrelationId() != null) ?
message.getHeaders().getCorrelationId(): message.getHeaders().getId();
message.getHeaders().getCorrelationId() : message.getHeaders().getId();
List<MessageBuilder<?>> messageBuilders = new ArrayList<MessageBuilder<?>>();
if (result instanceof Collection) {
Collection<?> items = (Collection<?>) result;
int sequenceNumber = 0;
int sequenceSize = items.size();
for (Object item : items) {
this.addReply(replyHolder, item, correlationId, ++sequenceNumber, sequenceSize);
messageBuilders.add(this.createBuilder(item, correlationId, ++sequenceNumber, sequenceSize));
}
}
else if (result.getClass().isArray()) {
@@ -52,19 +55,24 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess
int sequenceNumber = 0;
int sequenceSize = items.length;
for (Object item : items) {
this.addReply(replyHolder, item, correlationId, ++sequenceNumber, sequenceSize);
messageBuilders.add(this.createBuilder(item, correlationId, ++sequenceNumber, sequenceSize));
}
}
else {
this.addReply(replyHolder, result, correlationId, 1, 1);
messageBuilders.add(this.createBuilder(result, correlationId, 1, 1));
}
return messageBuilders;
}
private void addReply(ReplyMessageHolder replyHolder, Object item, Object correlationId, int sequenceNumber, int sequenceSize) {
replyHolder.add(item).setCorrelationId(correlationId)
@SuppressWarnings("unchecked")
private MessageBuilder createBuilder(Object item, Object correlationId, int sequenceNumber, int sequenceSize) {
MessageBuilder builder = (item instanceof Message) ?
MessageBuilder.fromMessage((Message) item) : MessageBuilder.withPayload(item);
builder.setCorrelationId(correlationId)
.setSequenceNumber(sequenceNumber)
.setSequenceSize(sequenceSize)
.setHeader(MessageHeaders.ID, UUID.randomUUID());
return builder;
}
/**

8
org.springframework.integration/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java
View File

@@ -18,7 +18,6 @@ package org.springframework.integration.transformer;
import org.springframework.integration.core.Message;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.ReplyMessageHolder;
import org.springframework.util.Assert;
/**
@@ -44,12 +43,9 @@ public class MessageTransformingHandler extends AbstractReplyProducingMessageHan
@Override
protected void handleRequestMessage(Message<?> message, ReplyMessageHolder replyHolder) {
protected Object handleRequestMessage(Message<?> message) {
try {
Message<?> result = transformer.transform(message);
if (result != null) {
replyHolder.set(result);
}
return transformer.transform(message);
}
catch (Exception e) {
if (e instanceof MessageTransformationException) {

43
org.springframework.integration/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java
View File

@@ -16,8 +16,16 @@
package org.springframework.integration.bus;
import static org.junit.Assert.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.channel.PublishSubscribeChannel;
@@ -28,15 +36,15 @@ import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.ReplyMessageHolder;
import org.springframework.integration.message.*;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.test.util.TestUtils.TestApplicationContext;
import org.springframework.scheduling.support.PeriodicTrigger;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
/**
* @author Mark Fisher
*/
@@ -53,8 +61,8 @@ public class ApplicationContextMessageBusTests {
.setReplyChannelName("targetChannel").build();
sourceChannel.send(message);
AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() {
public void handleRequestMessage(Message<?> message, ReplyMessageHolder replyHolder) {
replyHolder.set(message);
public Object handleRequestMessage(Message<?> message) {
return message;
}
};
handler.setBeanFactory(context);
@@ -99,14 +107,14 @@ public class ApplicationContextMessageBusTests {
QueueChannel outputChannel2 = new QueueChannel();
AbstractReplyProducingMessageHandler handler1 = new AbstractReplyProducingMessageHandler() {
@Override
public void handleRequestMessage(Message<?> message, ReplyMessageHolder replyHolder) {
replyHolder.set(message);
public Object handleRequestMessage(Message<?> message) {
return message;
}
};
AbstractReplyProducingMessageHandler handler2 = new AbstractReplyProducingMessageHandler() {
@Override
public void handleRequestMessage(Message<?> message, ReplyMessageHolder replyHolder) {
replyHolder.set(message);
public Object handleRequestMessage(Message<?> message) {
return message;
}
};
context.registerChannel("input", inputChannel);
@@ -135,16 +143,16 @@ public class ApplicationContextMessageBusTests {
final CountDownLatch latch = new CountDownLatch(2);
AbstractReplyProducingMessageHandler handler1 = new AbstractReplyProducingMessageHandler() {
@Override
public void handleRequestMessage(Message<?> message, ReplyMessageHolder replyHolder) {
replyHolder.set(message);
public Object handleRequestMessage(Message<?> message) {
latch.countDown();
return message;
}
};
AbstractReplyProducingMessageHandler handler2 = new AbstractReplyProducingMessageHandler() {
@Override
public void handleRequestMessage(Message<?> message, ReplyMessageHolder replyHolder) {
replyHolder.set(message);
public Object handleRequestMessage(Message<?> message) {
latch.countDown();
return message;
}
};
context.registerChannel("input", inputChannel);
@@ -198,8 +206,9 @@ public class ApplicationContextMessageBusTests {
final CountDownLatch latch = new CountDownLatch(1);
AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() {
@Override
public void handleRequestMessage(Message<?> message, ReplyMessageHolder replyHolder) {
public Object handleRequestMessage(Message<?> message) {
latch.countDown();
return null;
}
};
PollingConsumer endpoint = new PollingConsumer(errorChannel, handler);

5
org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java
View File

@@ -32,7 +32,6 @@ import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.ReplyMessageHolder;
import org.springframework.integration.handler.ServiceActivatingHandler;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.test.util.TestUtils;
@@ -89,7 +88,7 @@ public class DirectChannelSubscriptionTests {
public void exceptionThrownFromRegisteredEndpoint() {
AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() {
@Override
public void handleRequestMessage(Message<?> message, ReplyMessageHolder replyHolder) {
public Object handleRequestMessage(Message<?> message) {
throw new RuntimeException("intentional test failure");
}
};
@@ -124,7 +123,7 @@ public class DirectChannelSubscriptionTests {
}
private static class TestBean {
static class TestBean {
public Message<?> handle(Message<?> message) {
return new StringMessage(message.getPayload() + "!");

5
org.springframework.integration/src/test/java/org/springframework/integration/channel/MessageChannelTemplateTests.java
View File

@@ -34,7 +34,6 @@ import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.ReplyMessageHolder;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.test.util.TestUtils;
@@ -57,8 +56,8 @@ public class MessageChannelTemplateTests {
context.registerChannel("requestChannel", requestChannel);
AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() {
@Override
public void handleRequestMessage(Message<?> message, ReplyMessageHolder replyHolder) {
replyHolder.set(message.getPayload().toString().toUpperCase());
public Object handleRequestMessage(Message<?> message) {
return message.getPayload().toString().toUpperCase();
}
};
PollingConsumer endpoint = new PollingConsumer(requestChannel, handler);

5
org.springframework.integration/src/test/java/org/springframework/integration/config/ChainParserTests.java
View File

@@ -31,7 +31,6 @@ import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.ReplyMessageHolder;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -149,8 +148,8 @@ public class ChainParserTests {
public static class StubHandler extends AbstractReplyProducingMessageHandler {
@Override
protected void handleRequestMessage(Message<?> requestMessage, ReplyMessageHolder replyMessageHolder) {
replyMessageHolder.add(successMessage);
protected Object handleRequestMessage(Message<?> requestMessage) {
return successMessage;
}
}

4
org.springframework.integration/src/test/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandlerTest.java
View File

@@ -35,8 +35,8 @@ public class AbstractReplyProducingMessageHandlerTest {
private AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() {
@Override
protected void handleRequestMessage(Message<?> requestMessage, ReplyMessageHolder replyMessageHolder) {
replyMessageHolder.add(requestMessage);
protected Object handleRequestMessage(Message<?> requestMessage) {
return requestMessage;
}
};
private Message<?> message = MessageBuilder.withPayload("test").build();