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

Refactored SimpleMessagingGateway to include a base AbstractMessagingGateway with abstract methods for mapping messages to and from Objects. SimpleMessagingGateway delegates to a MessageMapper instance or uses DefaultMessageMapper.

Browse Source
This commit is contained in:
Mark Fisher
2008-09-25 21:18:01 +00:00
parent 6b659523b9
commit b0a00c7c0d
6 changed files with 212 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files

179
org.springframework.integration/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java Normal file
View File

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

20
org.springframework.integration/src/main/java/org/springframework/integration/gateway/DefaultMessageMapper.java
View File

@@ -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<T> implements MessageMapper<T> {
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;
}

6
org.springframework.integration/src/main/java/org/springframework/integration/gateway/MessageMapper.java
View File

@@ -23,10 +23,10 @@ import org.springframework.integration.message.Message;
*
* @author Mark Fisher
*/
public interface MessageMapper<T> {
public interface MessageMapper {
Message<?> toMessage(T object);
Message<?> toMessage(Object object);
T fromMessage(Message<?> message);
Object fromMessage(Message<?> message);
}

164
org.springframework.integration/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java
View File

@@ -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.
*
* <p>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<Object>();
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);
}
}

3
org.springframework.integration/src/test/java/org/springframework/integration/gateway/SimpleMessagingGatewayTests.java
View File

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

6
org.springframework.integration/src/test/java/org/springframework/integration/gateway/config/TestMessageMapper.java
View File

@@ -23,13 +23,13 @@ import org.springframework.integration.message.StringMessage;
/**
* @author Mark Fisher
*/
public class TestMessageMapper implements MessageMapper<String> {
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";
}