diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java new file mode 100644 index 0000000000..b6a351f60d --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java @@ -0,0 +1,179 @@ +/* + * 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.gateway; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.integration.ConfigurationException; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.bus.MessageBusAware; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.PollableChannel; +import org.springframework.integration.endpoint.MessagingGateway; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageDeliveryException; +import org.springframework.util.Assert; + +/** + * A convenient base class for connecting application code to + * {@link MessageChannel}s for sending, receiving, or request-reply operations. + * Exposes setters for configuring request and reply {@link MessageChannel}s as + * well as the timeout values for sending and receiving Messages. + * + * @author Mark Fisher + */ +public abstract class AbstractMessagingGateway implements MessagingGateway, MessageBusAware, InitializingBean { + + private volatile MessageChannel requestChannel; + + private volatile MessageChannel replyChannel; + + private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate(); + + private volatile ReplyMessageCorrelator replyMessageCorrelator; + + private volatile MessageBus messageBus; + + private final Object replyMessageCorrelatorMonitor = new Object(); + + + + /** + * Set the request channel. + * + * @param requestChannel the channel to which request messages will be sent + */ + public void setRequestChannel(MessageChannel requestChannel) { + this.requestChannel = requestChannel; + } + + /** + * Set the reply channel. If no reply channel is provided, this template will + * always use an anonymous, temporary channel for handling replies. + * + * @param replyChannel the channel from which reply messages will be received + */ + public void setReplyChannel(PollableChannel replyChannel) { + this.replyChannel = replyChannel; + } + + /** + * Set the timeout value for sending request messages. If not + * explicitly configured, the default is an indefinite timeout. + * + * @param requestTimeout the timeout value in milliseconds + */ + public void setRequestTimeout(long requestTimeout) { + this.channelTemplate.setSendTimeout(requestTimeout); + } + + /** + * Set the timeout value for receiving reply messages. If not + * explicitly configured, the default is an indefinite timeout. + * + * @param replyTimeout the timeout value in milliseconds + */ + public void setReplyTimeout(long replyTimeout) { + this.channelTemplate.setReceiveTimeout(replyTimeout); + } + + public void setMessageBus(MessageBus messageBus) { + this.messageBus = messageBus; + } + + public void afterPropertiesSet() throws Exception { + Assert.notNull(this.requestChannel, "requestChannel must not be null"); + } + + public void send(Object object) { + if (this.requestChannel == null) { + throw new IllegalStateException( + "send is not supported, because no request channel has been configured"); + } + Message message = this.toMessage(object); + Assert.notNull(message, "message must not be null"); + if (!this.channelTemplate.send(message, this.requestChannel)) { + throw new MessageDeliveryException(message, "failed to send Message to channel"); + } + } + + public Object receive() { + if (this.replyChannel == null || !(this.replyChannel instanceof PollableChannel)) { + throw new IllegalStateException( + "no-arg receive is not supported, because no pollable reply channel has been configured"); + } + Message message = this.channelTemplate.receive((PollableChannel) this.replyChannel); + return this.fromMessage(message); + } + + public Object sendAndReceive(Object object) { + return this.sendAndReceive(object, true); + } + + public Message sendAndReceiveMessage(Object object) { + return (Message) this.sendAndReceive(object, false); + } + + private Object sendAndReceive(Object object, boolean shouldMapMessage) { + Message request = this.toMessage(object); + Message reply = this.sendAndReceiveMessage(request); + if (!shouldMapMessage) { + return reply; + } + return this.fromMessage(reply); + } + + private Message sendAndReceiveMessage(Message message) { + Assert.notNull(message, "request message must not be null"); + if (this.requestChannel == null) { + throw new MessageDeliveryException(message, + "No request channel available. Cannot send request message."); + } + if (this.replyChannel != null && this.replyMessageCorrelator == null) { + this.registerReplyMessageCorrelator(); + } + return this.channelTemplate.sendAndReceive(message, this.requestChannel); + } + + private void registerReplyMessageCorrelator() { + synchronized (this.replyMessageCorrelatorMonitor) { + if (this.replyMessageCorrelator != null) { + return; + } + if (this.messageBus == null) { + throw new ConfigurationException("No MessageBus available. Cannot register ReplyMessageCorrelator."); + } + ReplyMessageCorrelator correlator = new ReplyMessageCorrelator(); + correlator.setBeanName("internal.correlator." + this); + correlator.setInputChannel(this.replyChannel); + correlator.afterPropertiesSet(); + this.messageBus.registerEndpoint(correlator); + this.replyMessageCorrelator = correlator; + } + } + + /** + * Subclasses must implement this to map from an Object to a Message. + */ + protected abstract Message toMessage(Object object); + + /** + * Subclasses must implement this to map from a Message to an Object. + */ + protected abstract Object fromMessage(Message message); + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/DefaultMessageMapper.java b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/DefaultMessageMapper.java index a1cda503ff..2a7fce2bdb 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/DefaultMessageMapper.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/DefaultMessageMapper.java @@ -20,22 +20,28 @@ import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageBuilder; /** - * A default implementation of both the {@link MessageMapper} and - * {@link OutboundMessageMapper} strategy interfaces. + * A default implementation of the {@link MessageMapper} strategy interface. * * @author Mark Fisher */ -public class DefaultMessageMapper implements MessageMapper { +public class DefaultMessageMapper implements MessageMapper { - @SuppressWarnings("unchecked") - public T fromMessage(Message message) { + /** + * Returns the Message payload (or null if the Message is null). + */ + public Object fromMessage(Message message) { if (message == null || message.getPayload() == null) { return null; } - return (T) message.getPayload(); + return message.getPayload(); } - public Message toMessage(T object) { + /** + * Returns a Message with the given object as its payload, unless the + * object is already a Message in which case it will be returned as-is. + * If the object is null, the returned Message will also be null. + */ + public Message toMessage(Object object) { if (object == null) { return null; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/MessageMapper.java b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/MessageMapper.java index 03f6c1969d..31b9d48ffc 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/MessageMapper.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/MessageMapper.java @@ -23,10 +23,10 @@ import org.springframework.integration.message.Message; * * @author Mark Fisher */ -public interface MessageMapper { +public interface MessageMapper { - Message toMessage(T object); + Message toMessage(Object object); - T fromMessage(Message message); + Object fromMessage(Message message); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java index 9040e5ffc1..9a352d6e63 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java @@ -16,176 +16,36 @@ package org.springframework.integration.gateway; -import org.springframework.beans.factory.InitializingBean; -import org.springframework.integration.ConfigurationException; -import org.springframework.integration.bus.MessageBus; -import org.springframework.integration.bus.MessageBusAware; -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.MessageChannelTemplate; -import org.springframework.integration.channel.PollableChannel; -import org.springframework.integration.endpoint.MessagingGateway; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageDeliveryException; import org.springframework.util.Assert; /** - * A convenient base class for connecting application code to - * {@link MessageChannel}s for sending, receiving, or request-reply operations. - * Exposes setters for configuring request and reply {@link MessageChannel}s as - * well as the timeout values for sending and receiving Messages. - * - *

By default, each request Message will be created with the method - * parameter as its payload, and each reply Message's payload will be the - * return value. To provide custom behavior for object-to-request and/or - * reply-to-object conversion, implement and set a 'messageMapper'. + * An implementation of {@link AbstractMessagingGateway} that delegates to + * a {@link MessageMapper}. The default is {@link DefaultMessageMapper}. * * @see MessageMapper * * @author Mark Fisher */ -public class SimpleMessagingGateway implements MessagingGateway, MessageBusAware, InitializingBean { +public class SimpleMessagingGateway extends AbstractMessagingGateway { - private volatile MessageChannel requestChannel; - - private volatile MessageChannel replyChannel; - - private volatile MessageMapper messageMapper = new DefaultMessageMapper(); - - private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate(); - - private volatile ReplyMessageCorrelator replyMessageCorrelator; - - private volatile MessageBus messageBus; - - private final Object replyMessageCorrelatorMonitor = new Object(); + private volatile MessageMapper messageMapper = new DefaultMessageMapper(); - public SimpleMessagingGateway(MessageChannel requestChannel) { - this.requestChannel = requestChannel; - } - - public SimpleMessagingGateway() { - super(); - } - - - /** - * Set the request channel. - * - * @param requestChannel the channel to which request messages will be sent - */ - public void setRequestChannel(MessageChannel requestChannel) { - this.requestChannel = requestChannel; - } - - /** - * Set the reply channel. If no reply channel is provided, this template will - * always use an anonymous, temporary channel for handling replies. - * - * @param replyChannel the channel from which reply messages will be received - */ - public void setReplyChannel(PollableChannel replyChannel) { - this.replyChannel = replyChannel; - } - - /** - * Set the timeout value for sending request messages. If not - * explicitly configured, the default is an indefinite timeout. - * - * @param requestTimeout the timeout value in milliseconds - */ - public void setRequestTimeout(long requestTimeout) { - this.channelTemplate.setSendTimeout(requestTimeout); - } - - /** - * Set the timeout value for receiving reply messages. If not - * explicitly configured, the default is an indefinite timeout. - * - * @param replyTimeout the timeout value in milliseconds - */ - public void setReplyTimeout(long replyTimeout) { - this.channelTemplate.setReceiveTimeout(replyTimeout); - } - - public void setMessageMapper(MessageMapper messageMapper) { + public void setMessageMapper(MessageMapper messageMapper) { Assert.notNull(messageMapper, "messageMapper must not be null"); - this.messageMapper = messageMapper; + this.messageMapper = (messageMapper != null) + ? messageMapper : new DefaultMessageMapper(); } - public void setMessageBus(MessageBus messageBus) { - this.messageBus = messageBus; - } - - public void afterPropertiesSet() throws Exception { - Assert.notNull(this.requestChannel, "requestChannel must not be null"); - } - - public void send(Object object) { - if (this.requestChannel == null) { - throw new IllegalStateException( - "send is not supported, because no request channel has been configured"); - } - Message message = this.messageMapper.toMessage(object); - Assert.notNull(message, "message must not be null"); - if (!this.channelTemplate.send(message, this.requestChannel)) { - throw new MessageDeliveryException(message, "failed to send Message to channel"); - } - } - - public Object receive() { - if (this.replyChannel == null || !(this.replyChannel instanceof PollableChannel)) { - throw new IllegalStateException( - "no-arg receive is not supported, because no pollable reply channel has been configured"); - } - Message message = this.channelTemplate.receive((PollableChannel) this.replyChannel); + @Override + protected Object fromMessage(Message message) { return this.messageMapper.fromMessage(message); } - public Object sendAndReceive(Object object) { - return this.sendAndReceive(object, true); - } - - public Message sendAndReceiveMessage(Object object) { - return (Message) this.sendAndReceive(object, false); - } - - private Object sendAndReceive(Object object, boolean shouldMapMessage) { - Message request = this.messageMapper.toMessage(object); - Message reply = this.sendAndReceiveMessage(request); - if (!shouldMapMessage) { - return reply; - } - return this.messageMapper.fromMessage(reply); - } - - private Message sendAndReceiveMessage(Message message) { - Assert.notNull(message, "request message must not be null"); - if (this.requestChannel == null) { - throw new MessageDeliveryException(message, - "No request channel available. Cannot send request message."); - } - if (this.replyChannel != null && this.replyMessageCorrelator == null) { - this.registerReplyMessageCorrelator(); - } - return this.channelTemplate.sendAndReceive(message, this.requestChannel); - } - - private void registerReplyMessageCorrelator() { - synchronized (this.replyMessageCorrelatorMonitor) { - if (this.replyMessageCorrelator != null) { - return; - } - if (this.messageBus == null) { - throw new ConfigurationException("No MessageBus available. Cannot register ReplyMessageCorrelator."); - } - ReplyMessageCorrelator correlator = new ReplyMessageCorrelator(); - correlator.setBeanName("internal.correlator." + this); - correlator.setInputChannel(this.replyChannel); - correlator.afterPropertiesSet(); - this.messageBus.registerEndpoint(correlator); - this.replyMessageCorrelator = correlator; - } + @Override + protected Message toMessage(Object object) { + return this.messageMapper.toMessage(object); } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/gateway/SimpleMessagingGatewayTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/gateway/SimpleMessagingGatewayTests.java index 2531ede0f8..2c9e18d9c2 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/gateway/SimpleMessagingGatewayTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/gateway/SimpleMessagingGatewayTests.java @@ -59,7 +59,8 @@ public class SimpleMessagingGatewayTests { @Before public void initializeSample() { - this.simpleMessagingGateway = new SimpleMessagingGateway(requestChannel); + this.simpleMessagingGateway = new SimpleMessagingGateway(); + this.simpleMessagingGateway.setRequestChannel(requestChannel); this.simpleMessagingGateway.setReplyChannel(replyChannel); this.simpleMessagingGateway.setMessageBus(messageBusMock); reset(allmocks); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/gateway/config/TestMessageMapper.java b/org.springframework.integration/src/test/java/org/springframework/integration/gateway/config/TestMessageMapper.java index b216bff50d..628b9ba06b 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/gateway/config/TestMessageMapper.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/gateway/config/TestMessageMapper.java @@ -23,13 +23,13 @@ import org.springframework.integration.message.StringMessage; /** * @author Mark Fisher */ -public class TestMessageMapper implements MessageMapper { +public class TestMessageMapper implements MessageMapper { - public Message toMessage(String object) { + public Message toMessage(Object object) { return new StringMessage("pre." + object); } - public String fromMessage(Message message) { + public Object fromMessage(Message message) { return message.getPayload().toString() + ".post"; }