INT-1129, INT-1263, INT-1306 removed SimpleMessagingGateway, refactored AbstractMessagingGateway to support GatewayProxyFactoryBean. AbstractMessagingGateway now uses HistoryWritingMessagePostProcessor.

spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java

@@ -18,7 +18,6 @@ package org.springframework.integration.gateway;
 
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageChannel;
-import org.springframework.integration.MessageDeliveryException;
 import org.springframework.integration.MessagingException;
 import org.springframework.integration.core.MessageHandler;
 import org.springframework.integration.core.MessagingTemplate;
@@ -28,11 +27,13 @@ import org.springframework.integration.endpoint.AbstractEndpoint;
 import org.springframework.integration.endpoint.EventDrivenConsumer;
 import org.springframework.integration.endpoint.PollingConsumer;
 import org.springframework.integration.handler.BridgeHandler;
-import org.springframework.integration.history.MessageHistory;
+import org.springframework.integration.history.HistoryWritingMessagePostProcessor;
 import org.springframework.integration.history.TrackableComponent;
 import org.springframework.integration.mapping.InboundMessageMapper;
+import org.springframework.integration.mapping.OutboundMessageMapper;
 import org.springframework.integration.message.ErrorMessage;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.support.converter.SimpleMessageConverter;
 import org.springframework.scheduling.support.PeriodicTrigger;
 import org.springframework.util.Assert;
 
@@ -44,13 +45,10 @@ import org.springframework.util.Assert;
  * 
  * @author Mark Fisher
  */
-public abstract class AbstractMessagingGateway extends AbstractEndpoint implements TrackableComponent{
+public abstract class AbstractMessagingGateway extends AbstractEndpoint implements TrackableComponent {
 
 	private static final long DEFAULT_TIMEOUT = 1000L;
 
-	private volatile boolean shouldTrack = false;
-
-	private volatile InboundMessageMapper<Throwable> exceptionMapper;
 
 	private volatile MessageChannel requestChannel;
 
@@ -58,7 +56,16 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint implemen
 
 	private volatile long replyTimeout = DEFAULT_TIMEOUT;
 
-	private final MessagingTemplate messagingTemplate = new MessagingTemplate();
+	@SuppressWarnings("rawtypes")
+	private volatile InboundMessageMapper requestMapper = new DefaultRequestMapper();
+
+	private volatile InboundMessageMapper<Throwable> exceptionMapper;
+
+	private final SimpleMessageConverter messageConverter = new SimpleMessageConverter();
+
+	private final MessagingTemplate messagingTemplate;
+
+	private final HistoryWritingMessagePostProcessor historyWritingPostProcessor = new HistoryWritingMessagePostProcessor();
 
 	private volatile boolean shouldThrowErrors = true;
 
@@ -70,18 +77,13 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint implemen
 
 
 	public AbstractMessagingGateway() {
-		this.messagingTemplate.setSendTimeout(DEFAULT_TIMEOUT);
-		this.messagingTemplate.setReceiveTimeout(this.replyTimeout);
+		MessagingTemplate template = new MessagingTemplate();
+		template.setMessageConverter(this.messageConverter);
+		template.setSendTimeout(DEFAULT_TIMEOUT);
+		template.setReceiveTimeout(this.replyTimeout);
+		this.messagingTemplate = template;
 	}
 
-	@Override
-	public String getComponentType(){
-		return "gateway";
-	}
-	
-	public void setShouldTrack(boolean shouldTrack) {
-		this.shouldTrack = shouldTrack;
-	}
 
 	/**
 	 * Set the request channel.
@@ -124,13 +126,21 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint implemen
 	}
 
 	/**
-	 * Specify whether the Throwable payload of a received {@link ErrorMessage}
-	 * should be extracted and thrown from a send-and-receive operation.
-	 * Otherwise, the ErrorMessage would be returned just like any other
-	 * reply Message. The default is <code>true</code>.
+	 * Provide an {@link InboundMessageMapper} for creating request Messages
+	 * from any object passed in a send or sendAndReceive operation.
 	 */
-	public void setShouldThrowErrors(boolean shouldThrowErrors) {
-		this.shouldThrowErrors = shouldThrowErrors;
+	public void setRequestMapper(InboundMessageMapper<?> requestMapper) {
+		requestMapper = (requestMapper != null) ? requestMapper : new DefaultRequestMapper();
+		this.requestMapper = requestMapper;
+		this.messageConverter.setInboundMessageMapper(requestMapper);
+	}
+
+	/**
+	 * Provide an {@link OutboundMessageMapper} for mapping to objects from
+	 * any reply Messages received in receive or sendAndReceive operations.
+	 */
+	public void setReplyMapper(OutboundMessageMapper<?> replyMapper) {
+		this.messageConverter.setOutboundMessageMapper(replyMapper);
 	}
 
 	/**
@@ -143,8 +153,32 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint implemen
 		this.exceptionMapper = exceptionMapper;
 	}
 
+	/**
+	 * Specify whether the Throwable payload of a received {@link ErrorMessage}
+	 * should be extracted and thrown from a send-and-receive operation.
+	 * Otherwise, the ErrorMessage would be returned just like any other
+	 * reply Message. The default is <code>true</code>.
+	 */
+	public void setShouldThrowErrors(boolean shouldThrowErrors) {
+		this.shouldThrowErrors = shouldThrowErrors;
+	}
+
+	/**
+	 * Specify whether this gateway should be tracked in the Message History
+	 * of Messages that originate from its send or sendAndReceive operations.
+	 */
+	public void setShouldTrack(boolean shouldTrack) {
+		this.historyWritingPostProcessor.setShouldTrack(shouldTrack);
+	}
+
+	@Override
+	public String getComponentType() {
+		return "gateway";
+	}
+
 	@Override
 	protected void onInit() throws Exception {
+		this.historyWritingPostProcessor.setTrackableComponent(this);
 		this.initialized = true;
 	}
 
@@ -156,80 +190,66 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint implemen
 
 	protected void send(Object object) {
 		this.initializeIfNecessary();
+		Assert.notNull(object, "request must not be null");
 		Assert.state(this.requestChannel != null,
 				"send is not supported, because no request channel has been configured");
-		Message<?> message = this.toMessage(object);
-		if (this.shouldTrack) {
-			message = MessageHistory.write(message, this);
-		}
-		
-		Assert.notNull(message, "message must not be null");
-		this.messagingTemplate.send(this.requestChannel, message);
+		this.messagingTemplate.convertAndSend(this.requestChannel, object, this.historyWritingPostProcessor);
 	}
 
 	protected Object receive() {
 		this.initializeIfNecessary();
 		Assert.state(this.replyChannel != null && (this.replyChannel instanceof PollableChannel),
 				"receive is not supported, because no pollable reply channel has been configured");
-		Message<?> message = this.messagingTemplate.receive((PollableChannel) this.replyChannel);
-		try {
-			return this.fromMessage(message);
-		}
-		catch (Exception e) {
-			if (e instanceof RuntimeException) {
-				throw (RuntimeException) e;
-			}
-			else throw new MessagingException(message, e);
-		}
+		return this.messagingTemplate.receiveAndConvert((PollableChannel) this.replyChannel);
 	}
 
 	protected Object sendAndReceive(Object object) {
-		return this.sendAndReceive(object, true);
+		return this.doSendAndReceive(object, true);
 	}
 
 	protected Message<?> sendAndReceiveMessage(Object object) {
-		return (Message<?>) this.sendAndReceive(object, false);
+		return (Message<?>) this.doSendAndReceive(object, false);
 	}
 
-	Object sendAndReceive(Object object, boolean shouldMapMessage) {
-		Message<?> request = this.toMessage(object);
-		if (this.shouldTrack) {
-			request = MessageHistory.write(request, this);
-		}
-		Message<?> reply = this.sendAndReceiveMessage(request);
-		if (!shouldMapMessage) {
-			return reply;
-		}
-		return this.fromMessage(reply);
-	}
-
-	private Message<?> sendAndReceiveMessage(Message<?> message) {
+	@SuppressWarnings("unchecked")
+	private Object doSendAndReceive(Object object, boolean shouldConvert) {
 		this.initializeIfNecessary();
-		Assert.notNull(message, "request message must not be null");
+		Assert.notNull(object, "request must not be null");
 		if (this.requestChannel == null) {
-			throw new MessageDeliveryException(message,
-					"No request channel available. Cannot send request message.");
+			throw new MessagingException("No request channel available. Cannot send request message.");
 		}
 		if (this.replyChannel != null && this.replyMessageCorrelator == null) {
 			this.registerReplyMessageCorrelator();
 		}
-		Message<?> reply = null;
+		Object reply = null;
 		Throwable error = null;
 		try {
-			reply = this.messagingTemplate.sendAndReceive(this.requestChannel, message);
-			if (reply instanceof ErrorMessage) {
-				error = ((ErrorMessage) reply).getPayload();
+			if (shouldConvert) {
+				reply = this.messagingTemplate.convertSendAndReceive(this.requestChannel, object, this.historyWritingPostProcessor);
+				if (reply instanceof Throwable) {
+					error = (Throwable) reply;
+				}
+			}
+			else {
+				Message<?> requestMessage = (object instanceof Message<?>)
+						? (Message<?>) object : this.requestMapper.toMessage(object);
+				requestMessage = this.historyWritingPostProcessor.postProcessMessage(requestMessage);
+				reply = this.messagingTemplate.sendAndReceive(this.requestChannel, requestMessage);
+				if (reply instanceof ErrorMessage) {
+					error = ((ErrorMessage) reply).getPayload();
+				}
 			}
 		}
 		catch (Exception e) {
-			logger.warn("failure occurred in gateway sendAndReceive.", e);
+			logger.warn("failure occurred in gateway sendAndReceive", e);
 			error = e;
 		}
 
 		if (error != null && this.exceptionMapper != null) {
 			try {
 				// create a reply message from the error
-				return this.exceptionMapper.toMessage(error);
+				Message<?> errorMessage = this.exceptionMapper.toMessage(error);
+				return (shouldConvert) ? errorMessage.getPayload() : errorMessage;
 			}
 			catch (Exception e2) {
 				// ignore this, we'll handle the original error next 
@@ -271,19 +291,6 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint implemen
 		}
 	}
 
-	protected Object fromMessage(Message<?> message) {
-		return (message != null ? message.getPayload() : null);
-	}
-
-	protected Message<?> toMessage(Object object) {
-		if (object instanceof Message<?>) {
-			return (Message<?>) object;
-		}
-		else {
-			return MessageBuilder.withPayload(object).build();
-		}
-	}
-
 	@Override // guarded by super#lifecycleLock
 	protected void doStart() {
 		if (this.replyMessageCorrelator != null) {
@@ -297,4 +304,16 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint implemen
 			this.replyMessageCorrelator.stop();
 		}
 	}
+
+
+	private static class DefaultRequestMapper implements InboundMessageMapper<Object> {
+
+		public Message<?> toMessage(Object object) throws Exception {
+			if (object instanceof Message<?>) {
+				return (Message<?>) object;
+			}
+			return (object != null) ? MessageBuilder.withPayload(object).build() : null;
+		}
+	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java

@@ -83,7 +83,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab
 
 	private volatile Object serviceProxy;
 
-	private final Map<Method, SimpleMessagingGateway> gatewayMap = new HashMap<Method, SimpleMessagingGateway>();
+	private final Map<Method, MethodInvocationGateway> gatewayMap = new HashMap<Method, MethodInvocationGateway>();
 
 	private volatile boolean initialized;
 
@@ -162,7 +162,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab
 	public void setShouldTrack(boolean shouldTrack) {
 		this.shouldTrack = shouldTrack;
 		if (!CollectionUtils.isEmpty(this.gatewayMap)) {
-			for (SimpleMessagingGateway gateway : this.gatewayMap.values()) {
+			for (MethodInvocationGateway gateway : this.gatewayMap.values()) {
 				gateway.setShouldTrack(shouldTrack);
 			}
 		}
@@ -190,7 +190,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab
 			Class<?> proxyInterface = this.determineServiceInterface();
 			Method[] methods = proxyInterface.getDeclaredMethods();
 			for (Method method : methods) {
-				SimpleMessagingGateway gateway = this.createGatewayForMethod(method);
+				MethodInvocationGateway gateway = this.createGatewayForMethod(method);
 				this.gatewayMap.put(method, gateway);
 			}
 			this.serviceProxy = new ProxyFactory(proxyInterface, this).getProxy(this.beanClassLoader);
@@ -243,7 +243,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab
 			this.afterPropertiesSet();
 		}
 		Method method = invocation.getMethod();
-		SimpleMessagingGateway gateway = this.gatewayMap.get(method);
+		MethodInvocationGateway gateway = this.gatewayMap.get(method);
 		Class<?> returnType = method.getReturnType();
 		boolean isReturnTypeMessage = Message.class.isAssignableFrom(returnType);
 		boolean shouldReply = returnType != void.class;
@@ -282,7 +282,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab
 		throw originalException;
 	}
 
-	private SimpleMessagingGateway createGatewayForMethod(Method method) {
+	private MethodInvocationGateway createGatewayForMethod(Method method) {
 		Gateway gatewayAnnotation = method.getAnnotation(Gateway.class);
 		MessageChannel requestChannel = this.defaultRequestChannel;
 		MessageChannel replyChannel = this.defaultReplyChannel;
@@ -330,7 +330,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab
 			messageMapper.setPayloadExpression(payloadExpression);
 		}
  		messageMapper.setBeanFactory(this.getBeanFactory());
-		SimpleMessagingGateway gateway = new SimpleMessagingGateway(messageMapper, new SimpleMessageMapper());
+ 		MethodInvocationGateway gateway = new MethodInvocationGateway(messageMapper);
 		gateway.setExceptionMapper(exceptionMapper);
 		if (this.getTaskScheduler() != null) {
 			gateway.setTaskScheduler(this.getTaskScheduler());
@@ -361,14 +361,14 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab
 
 	@Override // guarded by super#lifecycleLock
 	protected void doStart() {
-		for (SimpleMessagingGateway gateway : this.gatewayMap.values()) {
+		for (MethodInvocationGateway gateway : this.gatewayMap.values()) {
 			gateway.start();
 		}
 	}
 
 	@Override // guarded by super#lifecycleLock
 	protected void doStop() {
-		for (SimpleMessagingGateway gateway : this.gatewayMap.values()) {
+		for (MethodInvocationGateway gateway : this.gatewayMap.values()) {
 			gateway.stop();
 		}
 	}
@@ -392,8 +392,18 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab
 	private <T> T convert(Object source, Class<T> expectedReturnType) {
 		if (this.getConversionService() != null) {
 			return this.getConversionService().convert(source, expectedReturnType);
-		} else {
+		}
+		else {
 			return typeConverter.convertIfNecessary(source, expectedReturnType);
 		}
 	}
+
+
+	private static class MethodInvocationGateway extends AbstractMessagingGateway {
+
+		private MethodInvocationGateway(GatewayMethodInboundMessageMapper messageMapper) {
+			this.setRequestMapper(messageMapper);
+		}
+	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/gateway/RemotingInboundGatewaySupport.java

@@ -23,7 +23,7 @@ import org.springframework.integration.Message;
  * 
  * @author Mark Fisher
  */
-public abstract class RemotingInboundGatewaySupport extends SimpleMessagingGateway implements RequestReplyExchanger {
+public abstract class RemotingInboundGatewaySupport extends AbstractMessagingGateway implements RequestReplyExchanger {
 
 	private volatile boolean expectReply = true;
 

spring-integration-core/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java

@@ -1,92 +0,0 @@
-/*
- * Copyright 2002-2010 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.integration.Message;
-import org.springframework.integration.MessagingException;
-import org.springframework.integration.mapping.InboundMessageMapper;
-import org.springframework.integration.mapping.OutboundMessageMapper;
-import org.springframework.util.Assert;
-
-/**
- * An implementation of {@link AbstractMessagingGateway} that delegates to
- * an {@link InboundMessageMapper} and {@link OutboundMessageMapper}. The
- * default implementation for both is {@link SimpleMessageMapper}.
- * 
- * @see InboundMessageMapper
- * @see OutboundMessageMapper
- * 
- * @author Mark Fisher
- */
-@SuppressWarnings({"unchecked"})
-public class SimpleMessagingGateway extends AbstractMessagingGateway  {
-
-	private final InboundMessageMapper inboundMapper;
-
-	private final OutboundMessageMapper outboundMapper;
-
-	public SimpleMessagingGateway() {
-		SimpleMessageMapper mapper = new SimpleMessageMapper();
-		this.inboundMapper = mapper;
-		this.outboundMapper = mapper;
-	}
-
-	public SimpleMessagingGateway(InboundMessageMapper<?> inboundMapper, OutboundMessageMapper<?> outboundMapper) {
-		Assert.notNull(inboundMapper, "InboundMessageMapper must not be null");
-		Assert.notNull(outboundMapper, "OutboundMessageMapper must not be null");
-		this.inboundMapper = inboundMapper;
-		this.outboundMapper = outboundMapper;
-	}
-
-	public Message<?> sendAndReceiveMessage(Object object) {
-		return (Message<?>) super.sendAndReceive(object, false);
-	}
-	
-	public Object sendAndReceive(Object object) {
-		return super.sendAndReceive(object, true);
-	}
-
-
-	@Override
-	protected Object fromMessage(Message<?> message) {
-		try {
-			return this.outboundMapper.fromMessage(message);
-		}
-		catch (Exception e) {
-			if (e instanceof RuntimeException) {
-				throw (RuntimeException) e;
-			}
-			throw new MessagingException(message, e);
-		}
-	}
-
-	@Override
-	protected Message<?> toMessage(Object object) {
-		Message<?> message = null;
-		try {
-			message = this.inboundMapper.toMessage(object);
-		}
-		catch (Exception e) {
-			if (e instanceof RuntimeException) {
-				throw (RuntimeException) e;
-			}
-			throw new MessagingException("failed to create Message", e);
-		}
-		return message;
-	}
-
-}