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

INT-906 Now retaining reply channel and error channel headers when invoking sendAndReceive() on the MessageChannelTemplate.

Browse Source
This commit is contained in:
Mark Fisher
2009-12-07 00:21:40 +00:00
parent 9495ffaa2a
commit 10319788f3
2 changed files with 113 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
org.springframework.integration/src/main/java/org/springframework/integration/channel/MessageChannelTemplate.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.
@@ -26,6 +26,7 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.selector.MessageSelector;
@@ -240,6 +241,8 @@ public class MessageChannelTemplate implements InitializingBean {
}
private Message<?> doSendAndReceive(Message<?> request, MessageChannel channel) {
Object originalReplyChannelHeader = request.getHeaders().getReplyChannel();
Object originalErrorChannelHeader = request.getHeaders().getErrorChannel();
TemporaryReplyChannel replyChannel = new TemporaryReplyChannel(this.receiveTimeout);
request = MessageBuilder.fromMessage(request)
.setReplyChannel(replyChannel)
@@ -248,7 +251,14 @@ public class MessageChannelTemplate implements InitializingBean {
if (!this.doSend(request, channel)) {
throw new MessageDeliveryException(request, "failed to send message to channel");
}
return this.doReceive(replyChannel);
Message<?> reply = this.doReceive(replyChannel);
if (reply != null) {
reply = MessageBuilder.fromMessage(reply)
.setHeader(MessageHeaders.REPLY_CHANNEL, originalReplyChannelHeader)
.setHeader(MessageHeaders.ERROR_CHANNEL, originalErrorChannelHeader)
.build();
}
return reply;
}
private MessageChannel getRequiredDefaultChannel() {

101
org.springframework.integration/src/test/java/org/springframework/integration/gateway/NestedGatewayTests.java Normal file
View File

@@ -0,0 +1,101 @@
/*
* 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.
* 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.gateway;
import static org.junit.Assert.assertEquals;
import org.junit.Test;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.ReplyMessageHolder;
import org.springframework.integration.message.MessageBuilder;
/**
* @author Mark Fisher
*/
public class NestedGatewayTests {
@Test
public void nestedWithinHandler() {
DirectChannel innerChannel = new DirectChannel();
DirectChannel outerChannel = new DirectChannel();
innerChannel.subscribe(new AbstractReplyProducingMessageHandler() {
@Override
protected void handleRequestMessage(Message<?> requestMessage, ReplyMessageHolder replyMessageHolder) {
replyMessageHolder.set(requestMessage.getPayload() + "-reply");
}
});
final SimpleMessagingGateway innerGateway = new SimpleMessagingGateway();
innerGateway.setRequestChannel(innerChannel);
innerGateway.afterPropertiesSet();
outerChannel.subscribe(new AbstractReplyProducingMessageHandler() {
@Override
protected void handleRequestMessage(Message<?> requestMessage, ReplyMessageHolder replyMessageHolder) {
replyMessageHolder.set(innerGateway.sendAndReceiveMessage(
"pre-" + requestMessage.getPayload()).getPayload() + "-post");
}
});
SimpleMessagingGateway outerGateway = new SimpleMessagingGateway();
outerGateway.setRequestChannel(outerChannel);
outerGateway.afterPropertiesSet();
Message<?> reply = outerGateway.sendAndReceiveMessage("test");
assertEquals("pre-test-reply-post", reply.getPayload());
}
@Test
public void replyChannelRetained() {
DirectChannel requestChannel = new DirectChannel();
DirectChannel replyChannel = new DirectChannel();
requestChannel.subscribe(new AbstractReplyProducingMessageHandler() {
@Override
protected void handleRequestMessage(Message<?> requestMessage, ReplyMessageHolder replyMessageHolder) {
replyMessageHolder.set(requestMessage.getPayload() + "-reply");
}
});
SimpleMessagingGateway gateway = new SimpleMessagingGateway();
gateway.setRequestChannel(requestChannel);
gateway.afterPropertiesSet();
Message<?> message = MessageBuilder.withPayload("test")
.setReplyChannel(replyChannel).build();
Message<?> reply = gateway.sendAndReceiveMessage(message);
assertEquals("test-reply", reply.getPayload());
assertEquals(replyChannel, reply.getHeaders().getReplyChannel());
}
@Test
public void errorChannelRetained() {
DirectChannel requestChannel = new DirectChannel();
DirectChannel errorChannel = new DirectChannel();
requestChannel.subscribe(new AbstractReplyProducingMessageHandler() {
@Override
protected void handleRequestMessage(Message<?> requestMessage, ReplyMessageHolder replyMessageHolder) {
replyMessageHolder.set(requestMessage.getPayload() + "-reply");
}
});
SimpleMessagingGateway gateway = new SimpleMessagingGateway();
gateway.setRequestChannel(requestChannel);
gateway.afterPropertiesSet();
Message<?> message = MessageBuilder.withPayload("test")
.setErrorChannel(errorChannel).build();
Message<?> reply = gateway.sendAndReceiveMessage(message);
assertEquals("test-reply", reply.getPayload());
assertEquals(errorChannel, reply.getHeaders().getErrorChannel());
}
}