diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/MessageChannelTemplate.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/MessageChannelTemplate.java index 81a90c3ea5..d8237f2bf8 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/MessageChannelTemplate.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/MessageChannelTemplate.java @@ -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; @@ -171,8 +172,8 @@ public class MessageChannelTemplate implements InitializingBean { public boolean send(final Message message, final MessageChannel channel) { TransactionTemplate txTemplate = this.getTransactionTemplate(); if (txTemplate != null) { - return (Boolean) txTemplate.execute(new TransactionCallback() { - public Object doInTransaction(TransactionStatus status) { + return txTemplate.execute(new TransactionCallback() { + public Boolean doInTransaction(TransactionStatus status) { return doSend(message, channel); } }); @@ -190,8 +191,8 @@ public class MessageChannelTemplate implements InitializingBean { public Message receive(final PollableChannel channel) { TransactionTemplate txTemplate = this.getTransactionTemplate(); if (txTemplate != null) { - return (Message) txTemplate.execute(new TransactionCallback() { - public Object doInTransaction(TransactionStatus status) { + return txTemplate.execute(new TransactionCallback>() { + public Message doInTransaction(TransactionStatus status) { return doReceive(channel); } }); @@ -206,8 +207,8 @@ public class MessageChannelTemplate implements InitializingBean { public Message sendAndReceive(final Message request, final MessageChannel channel) { TransactionTemplate txTemplate = this.getTransactionTemplate(); if (txTemplate != null) { - return (Message) txTemplate.execute(new TransactionCallback() { - public Object doInTransaction(TransactionStatus status) { + return txTemplate.execute(new TransactionCallback>() { + public Message doInTransaction(TransactionStatus status) { return doSendAndReceive(request, channel); } }); @@ -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() { diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/gateway/NestedGatewayTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/gateway/NestedGatewayTests.java new file mode 100644 index 0000000000..fa9a2aef16 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/gateway/NestedGatewayTests.java @@ -0,0 +1,100 @@ +/* + * 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.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 Object handleRequestMessage(Message requestMessage) { + return requestMessage.getPayload() + "-reply"; + } + }); + final SimpleMessagingGateway innerGateway = new SimpleMessagingGateway(); + innerGateway.setRequestChannel(innerChannel); + innerGateway.afterPropertiesSet(); + outerChannel.subscribe(new AbstractReplyProducingMessageHandler() { + @Override + protected Object handleRequestMessage(Message requestMessage) { + return 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 Object handleRequestMessage(Message requestMessage) { + return 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 Object handleRequestMessage(Message requestMessage) { + return 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()); + } + +}