DSL: add gateway() EIP-method

* Upgrade dependencies * Polishing `Transformers`
2014-05-22 20:43:38 +03:00
parent 77026e8261
commit 2a25a1960f
7 changed files with 368 additions and 118 deletions
--- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GatewayEndpointSpec.java
+++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GatewayEndpointSpec.java
@@ -0,0 +1,67 @@
+/*
+ * Copyright 2014 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.dsl;
+
+import org.springframework.integration.dsl.core.ConsumerEndpointSpec;
+import org.springframework.messaging.MessageChannel;
+
+/**
+ * @author Artem Bilan
+ */
+public final class GatewayEndpointSpec extends ConsumerEndpointSpec<GatewayEndpointSpec, GatewayMessageHandler> {
+
+	GatewayEndpointSpec(MessageChannel requestChannel) {
+		super(new GatewayMessageHandler());
+		this.target.getT2().setRequestChannel(requestChannel);
+	}
+
+	GatewayEndpointSpec(String requestChannel) {
+		super(new GatewayMessageHandler());
+		this.target.getT2().setRequestChannelName(requestChannel);
+	}
+
+	public GatewayEndpointSpec replyChannel(MessageChannel replyChannel) {
+		this.target.getT2().setReplyChannel(replyChannel);
+		return this;
+	}
+
+	public GatewayEndpointSpec replyChannel(String replyChannel) {
+		this.target.getT2().setReplyChannelName(replyChannel);
+		return this;
+	}
+
+	public GatewayEndpointSpec errorChannel(MessageChannel errorChannel) {
+		this.target.getT2().setErrorChannel(errorChannel);
+		return this;
+	}
+
+	public GatewayEndpointSpec errorChannel(String errorChannel) {
+		this.target.getT2().setErrorChannelName(errorChannel);
+		return this;
+	}
+
+	public GatewayEndpointSpec requestTimeout(Long requestTimeout) {
+		this.target.getT2().setRequestTimeout(requestTimeout);
+		return this;
+	}
+
+	public GatewayEndpointSpec replyTimeout(Long replyTimeout) {
+		this.target.getT2().setReplyTimeout(replyTimeout);
+		return this;
+	}
+
+}
--- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GatewayMessageHandler.java
+++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GatewayMessageHandler.java
@@ -0,0 +1,116 @@
+/*
+ * Copyright 2014 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.dsl;
+
+import org.springframework.beans.factory.BeanCreationException;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.integration.gateway.GatewayProxyFactoryBean;
+import org.springframework.integration.gateway.RequestReplyExchanger;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.util.StringUtils;
+
+/**
+ * @author Artem Bilan
+ */
+class GatewayMessageHandler extends AbstractReplyProducingMessageHandler {
+
+	private final GatewayProxyFactoryBean gatewayProxyFactoryBean;
+
+	private RequestReplyExchanger exchanger;
+
+	private String requestChannel;
+
+	private String replyChannel;
+
+	private String errorChannel;
+
+	GatewayMessageHandler() {
+		this.gatewayProxyFactoryBean = new GatewayProxyFactoryBean();
+		this.gatewayProxyFactoryBean.setServiceInterface(RequestReplyExchanger.class);
+	}
+
+	void setRequestChannel(MessageChannel requestChannel) {
+		this.gatewayProxyFactoryBean.setDefaultRequestChannel(requestChannel);
+	}
+
+	void setRequestChannelName(String requestChannel) {
+		this.requestChannel = requestChannel;
+	}
+
+	public void setReplyChannel(MessageChannel replyChannel) {
+		this.gatewayProxyFactoryBean.setDefaultReplyChannel(replyChannel);
+	}
+
+	public void setReplyChannelName(String replyChannel) {
+		this.replyChannel = replyChannel;
+	}
+
+	public void setErrorChannel(MessageChannel errorChannel) {
+		this.gatewayProxyFactoryBean.setErrorChannel(errorChannel);
+	}
+
+	public void setErrorChannelName(String errorChannel) {
+		this.errorChannel = errorChannel;
+	}
+
+	public void setRequestTimeout(Long requestTimeout) {
+		this.gatewayProxyFactoryBean.setDefaultRequestTimeout(requestTimeout);
+	}
+
+	public void setReplyTimeout(Long replyTimeout) {
+		this.gatewayProxyFactoryBean.setDefaultReplyTimeout(replyTimeout);
+	}
+
+	@Override
+	protected void doInit() {
+		BeanFactory beanFactory = getBeanFactory();
+
+		if (StringUtils.hasText(this.requestChannel)) {
+			this.gatewayProxyFactoryBean.setDefaultRequestChannel(beanFactory.getBean(this.requestChannel,
+					MessageChannel.class));
+		}
+
+		if (StringUtils.hasText(this.replyChannel)) {
+			this.gatewayProxyFactoryBean.setDefaultReplyChannel(beanFactory.getBean(this.replyChannel,
+					MessageChannel.class));
+		}
+
+		if (StringUtils.hasText(this.errorChannel)) {
+			this.gatewayProxyFactoryBean.setErrorChannel(beanFactory.getBean(this.errorChannel,
+					MessageChannel.class));
+		}
+
+		if (beanFactory instanceof ConfigurableListableBeanFactory) {
+			((ConfigurableListableBeanFactory) beanFactory).initializeBean(this.gatewayProxyFactoryBean, null);
+		}
+		try {
+			this.exchanger = (RequestReplyExchanger) this.gatewayProxyFactoryBean.getObject();
+		}
+		catch (Exception e) {
+			throw new BeanCreationException("Can't instantiate the GatewayProxyFactoryBean: " + this, e);
+		}
+	}
+
+	@Override
+	protected Object handleRequestMessage(Message<?> requestMessage) {
+		return exchanger.exchange(requestMessage);
+	}
+
+}
--- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java
+++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java
@@ -140,7 +140,7 @@ public final class IntegrationFlowBuilder {
 	}

 	public <S, T> IntegrationFlowBuilder transform(GenericTransformer<S, T> genericTransformer,
-												   EndpointConfigurer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
+			EndpointConfigurer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
 		Assert.notNull(genericTransformer);
 		Transformer transformer = genericTransformer instanceof Transformer
 				? (Transformer) genericTransformer : new MethodInvokingTransformer(genericTransformer);
@@ -173,13 +173,13 @@ public final class IntegrationFlowBuilder {
 	}

 	public IntegrationFlowBuilder handle(String beanName, String methodName,
-										 EndpointConfigurer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
+			EndpointConfigurer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
 		return this.handle(new ServiceActivatingHandler(new BeanNameMessageProcessor<Object>(beanName, methodName)),
 				endpointConfigurer);
 	}

 	public <H extends MessageHandler> IntegrationFlowBuilder handle(H messageHandler,
-																	EndpointConfigurer<GenericEndpointSpec<H>> endpointConfigurer) {
+			EndpointConfigurer<GenericEndpointSpec<H>> endpointConfigurer) {
 		Assert.notNull(messageHandler);
 		return this.register(new GenericEndpointSpec<H>(messageHandler), endpointConfigurer);
 	}
@@ -206,7 +206,7 @@ public final class IntegrationFlowBuilder {
 	}

 	public IntegrationFlowBuilder enrich(ComponentConfigurer<EnricherSpec> enricherConfigurer,
-										 EndpointConfigurer<GenericEndpointSpec<ContentEnricher>> endpointConfigurer) {
+			EndpointConfigurer<GenericEndpointSpec<ContentEnricher>> endpointConfigurer) {
 		Assert.notNull(enricherConfigurer);
 		EnricherSpec enricherSpec = new EnricherSpec();
 		enricherConfigurer.configure(enricherSpec);
@@ -227,7 +227,7 @@ public final class IntegrationFlowBuilder {
 	}

 	public IntegrationFlowBuilder enrichHeaders(ComponentConfigurer<HeaderEnricherSpec> headerEnricherConfigurer,
-												EndpointConfigurer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
+			EndpointConfigurer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
 		Assert.notNull(headerEnricherConfigurer);
 		HeaderEnricherSpec headerEnricherSpec = new HeaderEnricherSpec();
 		headerEnricherConfigurer.configure(headerEnricherSpec);
@@ -239,7 +239,7 @@ public final class IntegrationFlowBuilder {
 	}

 	public IntegrationFlowBuilder enrichHeaders(HeaderEnricher headerEnricher,
-												EndpointConfigurer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
+			EndpointConfigurer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
 		return this.addComponent(headerEnricher).transform(headerEnricher, endpointConfigurer);
 	}

@@ -247,7 +247,8 @@ public final class IntegrationFlowBuilder {
 		return this.split((EndpointConfigurer<SplitterEndpointSpec<DefaultMessageSplitter>>) null);
 	}

-	public IntegrationFlowBuilder split(EndpointConfigurer<SplitterEndpointSpec<DefaultMessageSplitter>> endpointConfigurer) {
+	public
+	IntegrationFlowBuilder split(EndpointConfigurer<SplitterEndpointSpec<DefaultMessageSplitter>> endpointConfigurer) {
 		return this.split(new DefaultMessageSplitter(), endpointConfigurer);
 	}

@@ -265,7 +266,7 @@ public final class IntegrationFlowBuilder {
 	}

 	public IntegrationFlowBuilder split(String beanName, String methodName,
-										EndpointConfigurer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
+			EndpointConfigurer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
 		return this.split(new MethodInvokingSplitter(new BeanNameMessageProcessor<Collection<?>>(beanName, methodName)),
 				endpointConfigurer);
 	}
@@ -279,7 +280,7 @@ public final class IntegrationFlowBuilder {
 	}

 	public <T> IntegrationFlowBuilder split(GenericSplitter<T> splitter,
-											EndpointConfigurer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
+			EndpointConfigurer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
 		return this.split(new MethodInvokingSplitter(splitter, "split"), endpointConfigurer);
 	}

@@ -291,7 +292,8 @@ public final class IntegrationFlowBuilder {

 	/**
 	 * Provides the {@link HeaderFilter} to the current {@link IntegrationFlow}.
-	 * @param headersToRemove the array of headers (or patterns) to remove from {@link org.springframework.messaging.MessageHeaders}.
+	 * @param headersToRemove the array of headers (or patterns)
+	 * to remove from {@link org.springframework.messaging.MessageHeaders}.
 	 * @return the {@link IntegrationFlowBuilder}.
 	 */
 	public IntegrationFlowBuilder headerFilter(String... headersToRemove) {
@@ -314,7 +316,7 @@ public final class IntegrationFlowBuilder {
 	}

 	public IntegrationFlowBuilder headerFilter(HeaderFilter headerFilter,
-											   EndpointConfigurer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
+			EndpointConfigurer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
 		return this.transform(headerFilter, endpointConfigurer);
 	}

@@ -336,7 +338,7 @@ public final class IntegrationFlowBuilder {
 	}

 	public IntegrationFlowBuilder claimCheckOut(MessageStore messageStore, boolean removeMessage,
-												EndpointConfigurer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
+			EndpointConfigurer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
 		ClaimCheckOutTransformer claimCheckOutTransformer = new ClaimCheckOutTransformer(messageStore);
 		claimCheckOutTransformer.setRemoveMessage(removeMessage);
 		return this.transform(claimCheckOutTransformer, endpointConfigurer);
@@ -346,8 +348,10 @@ public final class IntegrationFlowBuilder {
 		return this.resequence((EndpointConfigurer<GenericEndpointSpec<ResequencingMessageHandler>>) null);
 	}

-	public IntegrationFlowBuilder resequence(EndpointConfigurer<GenericEndpointSpec<ResequencingMessageHandler>> endpointConfigurer) {
-		return this.resequence(new ResequencingMessageHandler(new ResequencingMessageGroupProcessor()), endpointConfigurer);
+	public
+	IntegrationFlowBuilder resequence(EndpointConfigurer<GenericEndpointSpec<ResequencingMessageHandler>> endpointConfigurer) {
+		return this.resequence(new ResequencingMessageHandler(new ResequencingMessageGroupProcessor()),
+				endpointConfigurer);
 	}

 	public IntegrationFlowBuilder resequence(ComponentConfigurer<ResequencerSpec> resequencerConfigurer) {
@@ -355,7 +359,7 @@ public final class IntegrationFlowBuilder {
 	}

 	public IntegrationFlowBuilder resequence(ComponentConfigurer<ResequencerSpec> resequencerConfigurer,
-											 EndpointConfigurer<GenericEndpointSpec<ResequencingMessageHandler>> endpointConfigurer) {
+			EndpointConfigurer<GenericEndpointSpec<ResequencingMessageHandler>> endpointConfigurer) {
 		Assert.notNull(resequencerConfigurer);
 		ResequencerSpec spec = new ResequencerSpec();
 		resequencerConfigurer.configure(spec);
@@ -367,7 +371,7 @@ public final class IntegrationFlowBuilder {
 	}

 	public IntegrationFlowBuilder resequence(ResequencingMessageHandler resequencer,
-											 EndpointConfigurer<GenericEndpointSpec<ResequencingMessageHandler>> endpointConfigurer) {
+			EndpointConfigurer<GenericEndpointSpec<ResequencingMessageHandler>> endpointConfigurer) {
 		return this.handle(resequencer, endpointConfigurer);
 	}

@@ -375,7 +379,8 @@ public final class IntegrationFlowBuilder {
 		return this.aggregate((EndpointConfigurer<GenericEndpointSpec<AggregatingMessageHandler>>) null);
 	}

-	public IntegrationFlowBuilder aggregate(EndpointConfigurer<GenericEndpointSpec<AggregatingMessageHandler>> endpointConfigurer) {
+	public
+	IntegrationFlowBuilder aggregate(EndpointConfigurer<GenericEndpointSpec<AggregatingMessageHandler>> endpointConfigurer) {
 		return this.aggregate(new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor()),
 				endpointConfigurer);
 	}
@@ -385,7 +390,7 @@ public final class IntegrationFlowBuilder {
 	}

 	public IntegrationFlowBuilder aggregate(ComponentConfigurer<AggregatorSpec> aggregatorConfigurer,
-											EndpointConfigurer<GenericEndpointSpec<AggregatingMessageHandler>> endpointConfigurer) {
+			EndpointConfigurer<GenericEndpointSpec<AggregatingMessageHandler>> endpointConfigurer) {
 		Assert.notNull(aggregatorConfigurer);
 		AggregatorSpec spec = new AggregatorSpec();
 		aggregatorConfigurer.configure(spec);
@@ -397,7 +402,7 @@ public final class IntegrationFlowBuilder {
 	}

 	public IntegrationFlowBuilder aggregate(AggregatingMessageHandler aggregator,
-											EndpointConfigurer<GenericEndpointSpec<AggregatingMessageHandler>> endpointConfigurer) {
+			EndpointConfigurer<GenericEndpointSpec<AggregatingMessageHandler>> endpointConfigurer) {
 		return this.handle(aggregator, endpointConfigurer);
 	}

@@ -407,13 +412,13 @@ public final class IntegrationFlowBuilder {
 	}

 	public IntegrationFlowBuilder route(String beanName, String method,
-										ComponentConfigurer<RouterSpec<MethodInvokingRouter>> routerConfigurer) {
+			ComponentConfigurer<RouterSpec<MethodInvokingRouter>> routerConfigurer) {
 		return this.route(beanName, method, routerConfigurer, null);
 	}

 	public IntegrationFlowBuilder route(String beanName, String method,
-										ComponentConfigurer<RouterSpec<MethodInvokingRouter>> routerConfigurer,
-										EndpointConfigurer<GenericEndpointSpec<MethodInvokingRouter>> endpointConfigurer) {
+			ComponentConfigurer<RouterSpec<MethodInvokingRouter>> routerConfigurer,
+			EndpointConfigurer<GenericEndpointSpec<MethodInvokingRouter>> endpointConfigurer) {
 		return this.route(new MethodInvokingRouter(new BeanNameMessageProcessor<Object>(beanName, method)),
 				routerConfigurer, endpointConfigurer);
 	}
@@ -429,8 +434,8 @@ public final class IntegrationFlowBuilder {
 	}

 	public IntegrationFlowBuilder route(String expression,
-										ComponentConfigurer<RouterSpec<ExpressionEvaluatingRouter>> routerConfigurer,
-										EndpointConfigurer<GenericEndpointSpec<ExpressionEvaluatingRouter>> endpointConfigurer) {
+			ComponentConfigurer<RouterSpec<ExpressionEvaluatingRouter>> routerConfigurer,
+			EndpointConfigurer<GenericEndpointSpec<ExpressionEvaluatingRouter>> endpointConfigurer) {
 		return this.route(new ExpressionEvaluatingRouter(PARSER.parseExpression(expression)), routerConfigurer,
 				endpointConfigurer);
 	}
@@ -445,8 +450,8 @@ public final class IntegrationFlowBuilder {
 	}

 	public <S, T> IntegrationFlowBuilder route(GenericRouter<S, T> router,
-										ComponentConfigurer<RouterSpec<MethodInvokingRouter>> routerConfigurer,
-										EndpointConfigurer<GenericEndpointSpec<MethodInvokingRouter>> endpointConfigurer) {
+			ComponentConfigurer<RouterSpec<MethodInvokingRouter>> routerConfigurer,
+			EndpointConfigurer<GenericEndpointSpec<MethodInvokingRouter>> endpointConfigurer) {
 		return this.route(new MethodInvokingRouter(router), routerConfigurer, endpointConfigurer);
 	}

@@ -456,8 +461,8 @@ public final class IntegrationFlowBuilder {
 	}

 	public <R extends AbstractMappingMessageRouter> IntegrationFlowBuilder route(R router,
-																				 ComponentConfigurer<RouterSpec<R>> routerConfigurer,
-																				 EndpointConfigurer<GenericEndpointSpec<R>> endpointConfigurer) {
+			ComponentConfigurer<RouterSpec<R>> routerConfigurer,
+			EndpointConfigurer<GenericEndpointSpec<R>> endpointConfigurer) {
 		if (routerConfigurer != null) {
 			RouterSpec<R> routerSpec = new RouterSpec<R>(router);
 			routerConfigurer.configure(routerSpec);
@@ -470,7 +475,7 @@ public final class IntegrationFlowBuilder {
 	}

 	public IntegrationFlowBuilder recipientListRoute(ComponentConfigurer<RecipientListRouterSpec> routerConfigurer,
-													 EndpointConfigurer<GenericEndpointSpec<RecipientListRouter>> endpointConfigurer) {
+			EndpointConfigurer<GenericEndpointSpec<RecipientListRouter>> endpointConfigurer) {
 		Assert.notNull(routerConfigurer);
 		RecipientListRouterSpec spec = new RecipientListRouterSpec();
 		routerConfigurer.configure(spec);
@@ -484,10 +489,27 @@ public final class IntegrationFlowBuilder {
 	}

 	public <R extends AbstractMessageRouter> IntegrationFlowBuilder route(R router,
-																		  EndpointConfigurer<GenericEndpointSpec<R>> endpointConfigurer) {
+			EndpointConfigurer<GenericEndpointSpec<R>> endpointConfigurer) {
 		return this.handle(router, endpointConfigurer);
 	}

+	public IntegrationFlowBuilder gateway(String requestChannel) {
+		return gateway(requestChannel, null);
+	}
+
+	public IntegrationFlowBuilder gateway(String requestChannel,
+			EndpointConfigurer<GatewayEndpointSpec> endpointConfigurer) {
+		return register(new GatewayEndpointSpec(requestChannel), endpointConfigurer);
+	}
+
+	public IntegrationFlowBuilder gateway(MessageChannel requestChannel) {
+		return gateway(requestChannel, null);
+	}
+
+	public IntegrationFlowBuilder gateway(MessageChannel requestChannel,
+			EndpointConfigurer<GatewayEndpointSpec> endpointConfigurer) {
+		return register(new GatewayEndpointSpec(requestChannel), endpointConfigurer);
+	}

 	private <S extends ConsumerEndpointSpec<S, ?>> IntegrationFlowBuilder register(S endpointSpec,
 			EndpointConfigurer<S> endpointConfigurer) {
--- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/Amqp.java
+++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/Amqp.java
@@ -24,7 +24,6 @@ import org.springframework.integration.dsl.core.MessagingGatewaySpec;

 /**
 * @author Artem Bilan
- * @since 4.0
 */
 public abstract class Amqp {

--- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Transformers.java
+++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Transformers.java
@@ -28,8 +28,11 @@ import org.springframework.core.serializer.Serializer;
 import org.springframework.expression.Expression;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.dsl.tuple.Tuple2;
+import org.springframework.integration.file.transformer.FileToByteArrayTransformer;
+import org.springframework.integration.file.transformer.FileToStringTransformer;
 import org.springframework.integration.json.JsonToObjectTransformer;
 import org.springframework.integration.json.ObjectToJsonTransformer;
+import org.springframework.integration.mail.transformer.MailToStringTransformer;
 import org.springframework.integration.support.json.JsonObjectMapper;
 import org.springframework.integration.transformer.MapToObjectTransformer;
 import org.springframework.integration.transformer.ObjectToMapTransformer;
@@ -38,8 +41,19 @@ import org.springframework.integration.transformer.PayloadDeserializingTransform
 import org.springframework.integration.transformer.PayloadSerializingTransformer;
 import org.springframework.integration.transformer.PayloadTypeConvertingTransformer;
 import org.springframework.integration.transformer.SyslogToMapTransformer;
-import org.springframework.integration.transformer.Transformer;
+import org.springframework.integration.xml.result.ResultFactory;
+import org.springframework.integration.xml.source.SourceFactory;
+import org.springframework.integration.xml.transformer.MarshallingTransformer;
+import org.springframework.integration.xml.transformer.ResultTransformer;
+import org.springframework.integration.xml.transformer.SourceCreatingTransformer;
+import org.springframework.integration.xml.transformer.UnmarshallingTransformer;
+import org.springframework.integration.xml.transformer.XPathTransformer;
+import org.springframework.integration.xml.transformer.XsltPayloadTransformer;
+import org.springframework.integration.xml.xpath.XPathEvaluationType;
+import org.springframework.oxm.Marshaller;
+import org.springframework.oxm.Unmarshaller;
 import org.springframework.util.Assert;
+import org.springframework.xml.xpath.NodeMapper;

 /**
 * @author Artem Bilan
@@ -48,58 +62,58 @@ public abstract class Transformers {

 	private final static SpelExpressionParser PARSER = new SpelExpressionParser();

-	public static Transformer objectToString() {
+	public static ObjectToStringTransformer objectToString() {
 		return objectToString(null);
 	}

-	public static Transformer objectToString(String charset) {
+	public static ObjectToStringTransformer objectToString(String charset) {
 		return charset != null ? new ObjectToStringTransformer(charset) : new ObjectToStringTransformer();
 	}

-	public static Transformer toMap() {
+	public static ObjectToMapTransformer toMap() {
 		return new ObjectToMapTransformer();
 	}

-	public static Transformer toMap(boolean shouldFlattenKeys) {
+	public static ObjectToMapTransformer toMap(boolean shouldFlattenKeys) {
 		ObjectToMapTransformer transformer = new ObjectToMapTransformer();
 		transformer.setShouldFlattenKeys(shouldFlattenKeys);
 		return transformer;
 	}

-	public static Transformer fromMap(Class<?> targetClass) {
+	public static MapToObjectTransformer fromMap(Class<?> targetClass) {
 		return new MapToObjectTransformer(targetClass);
 	}

-	public static Transformer fromMap(String beanName) {
+	public static MapToObjectTransformer fromMap(String beanName) {
 		return new MapToObjectTransformer(beanName);
 	}

-	public static Transformer toJson() {
+	public static ObjectToJsonTransformer toJson() {
 		return toJson(null, null, null);
 	}

-	public static Transformer toJson(JsonObjectMapper<?, ?> jsonObjectMapper) {
+	public static ObjectToJsonTransformer toJson(JsonObjectMapper<?, ?> jsonObjectMapper) {
 		return toJson(jsonObjectMapper, null, null);
 	}

-	public static Transformer toJson(JsonObjectMapper<?, ?> jsonObjectMapper,
+	public static ObjectToJsonTransformer toJson(JsonObjectMapper<?, ?> jsonObjectMapper,
 			ObjectToJsonTransformer.ResultType resultType) {
 		return toJson(jsonObjectMapper, resultType, null);
 	}

-	public static Transformer toJson(String contentType) {
+	public static ObjectToJsonTransformer toJson(String contentType) {
 		return toJson(null, null, contentType);
 	}

-	public static Transformer toJson(JsonObjectMapper<?, ?> jsonObjectMapper, String contentType) {
+	public static ObjectToJsonTransformer toJson(JsonObjectMapper<?, ?> jsonObjectMapper, String contentType) {
 		return toJson(jsonObjectMapper, null, contentType);
 	}

-	public static Transformer toJson(ObjectToJsonTransformer.ResultType resultType, String contentType) {
+	public static ObjectToJsonTransformer toJson(ObjectToJsonTransformer.ResultType resultType, String contentType) {
 		return toJson(null, resultType, contentType);
 	}

-	public static Transformer toJson(JsonObjectMapper<?, ?> jsonObjectMapper,
+	public static ObjectToJsonTransformer toJson(JsonObjectMapper<?, ?> jsonObjectMapper,
 			ObjectToJsonTransformer.ResultType resultType, String contentType) {
 		ObjectToJsonTransformer transformer;
 		if (jsonObjectMapper != null) {
@@ -122,27 +136,27 @@ public abstract class Transformers {
 		return transformer;
 	}

-	public static Transformer fromJson() {
+	public static JsonToObjectTransformer fromJson() {
 		return fromJson(null, null);
 	}

-	public static Transformer fromJson(Class<?> targetClass) {
+	public static JsonToObjectTransformer fromJson(Class<?> targetClass) {
 		return fromJson(targetClass, null);
 	}

-	public static Transformer fromJson(JsonObjectMapper<?, ?> jsonObjectMapper) {
+	public static JsonToObjectTransformer fromJson(JsonObjectMapper<?, ?> jsonObjectMapper) {
 		return fromJson(null, jsonObjectMapper);
 	}

-	public static Transformer fromJson(Class<?> targetClass, JsonObjectMapper<?, ?> jsonObjectMapper) {
+	public static JsonToObjectTransformer fromJson(Class<?> targetClass, JsonObjectMapper<?, ?> jsonObjectMapper) {
 		return new JsonToObjectTransformer(targetClass, jsonObjectMapper);
 	}

-	public static Transformer serializer() {
+	public static PayloadSerializingTransformer serializer() {
 		return serializer(null);
 	}

-	public static Transformer serializer(Serializer<Object> serializer) {
+	public static PayloadSerializingTransformer serializer(Serializer<Object> serializer) {
 		PayloadSerializingTransformer transformer = new PayloadSerializingTransformer();
 		if (serializer != null) {
 			transformer.setSerializer(serializer);
@@ -150,11 +164,11 @@ public abstract class Transformers {
 		return transformer;
 	}

-	public static Transformer deserializer() {
+	public static PayloadDeserializingTransformer deserializer() {
 		return deserializer(null);
 	}

-	public static Transformer deserializer(Deserializer<Object> deserializer) {
+	public static PayloadDeserializingTransformer deserializer(Deserializer<Object> deserializer) {
 		PayloadDeserializingTransformer transformer = new PayloadDeserializingTransformer();
 		if (deserializer != null) {
 			transformer.setDeserializer(deserializer);
@@ -162,98 +176,86 @@ public abstract class Transformers {
 		return transformer;
 	}

-	public static <T, U> Transformer converter(Converter<T, U> converter) {
+	public static <T, U> PayloadTypeConvertingTransformer<T, U> converter(Converter<T, U> converter) {
 		Assert.notNull(converter, "The Converter<?, ?> is required for the PayloadTypeConvertingTransformer");
 		PayloadTypeConvertingTransformer<T, U> transformer = new PayloadTypeConvertingTransformer<T, U>();
 		transformer.setConverter(converter);
 		return transformer;
 	}

-	public static Transformer syslogToMap() {
+	public static SyslogToMapTransformer syslogToMap() {
 		return new SyslogToMapTransformer();
 	}

-	public static Transformer fromMail() {
+	public static MailToStringTransformer fromMail() {
 		return fromMail(null);
 	}

-	public static Transformer fromMail(String charset) {
-		org.springframework.integration.mail.transformer.MailToStringTransformer transformer =
-				new org.springframework.integration.mail.transformer.MailToStringTransformer();
+	public static MailToStringTransformer fromMail(String charset) {
+		MailToStringTransformer transformer = new MailToStringTransformer();
 		if (charset != null) {
 			transformer.setCharset(charset);
 		}
 		return transformer;
 	}

-	public static Transformer fileToString() {
+	public static FileToStringTransformer fileToString() {
 		return fileToString(null);
 	}

-	public static Transformer fileToString(String charset) {
-		org.springframework.integration.file.transformer.FileToStringTransformer transformer =
-				new org.springframework.integration.file.transformer.FileToStringTransformer();
+	public static FileToStringTransformer fileToString(String charset) {
+		FileToStringTransformer transformer = new FileToStringTransformer();
 		if (charset != null) {
 			transformer.setCharset(charset);
 		}
 		return transformer;
 	}

-	public static Transformer fileToByteArray() {
-		return new org.springframework.integration.file.transformer.FileToByteArrayTransformer();
+	public static FileToByteArrayTransformer fileToByteArray() {
+		return new FileToByteArrayTransformer();
 	}

-	public static Transformer marshaller(org.springframework.oxm.Marshaller marshaller) {
+	public static MarshallingTransformer marshaller(Marshaller marshaller) {
 		return marshaller(marshaller, null, null, null);
 	}

-	public static Transformer marshaller(org.springframework.oxm.Marshaller marshaller,
-			org.springframework.integration.xml.transformer.ResultTransformer resultTransformer) {
+	public static MarshallingTransformer marshaller(Marshaller marshaller, ResultTransformer resultTransformer) {
 		return marshaller(marshaller, resultTransformer, null);
 	}

-	public static Transformer marshaller(org.springframework.oxm.Marshaller marshaller,
-			org.springframework.integration.xml.result.ResultFactory resultFactory) {
+	public static MarshallingTransformer marshaller(Marshaller marshaller, ResultFactory resultFactory) {
 		return marshaller(marshaller, null, resultFactory);
 	}

-	public static Transformer marshaller(org.springframework.oxm.Marshaller marshaller, boolean extractPayload) {
+	public static MarshallingTransformer marshaller(Marshaller marshaller, boolean extractPayload) {
 		return marshaller(marshaller, null, null, extractPayload);
 	}

-	public static Transformer marshaller(org.springframework.oxm.Marshaller marshaller,
-			org.springframework.integration.xml.result.ResultFactory resultFactory,
+	public static MarshallingTransformer marshaller(Marshaller marshaller,ResultFactory resultFactory,
 			boolean extractPayload) {
 		return marshaller(marshaller, null, resultFactory, extractPayload);
 	}

-	public static Transformer marshaller(org.springframework.oxm.Marshaller marshaller,
-			org.springframework.integration.xml.transformer.ResultTransformer resultTransformer,
+	public static MarshallingTransformer marshaller(Marshaller marshaller, ResultTransformer resultTransformer,
 			boolean extractPayload) {
 		return marshaller(marshaller, resultTransformer, null, extractPayload);
 	}


-	public static Transformer marshaller(org.springframework.oxm.Marshaller marshaller,
-			org.springframework.integration.xml.transformer.ResultTransformer resultTransformer,
-			org.springframework.integration.xml.result.ResultFactory resultFactory) {
+	public static MarshallingTransformer marshaller(Marshaller marshaller, ResultTransformer resultTransformer,
+			ResultFactory resultFactory) {
 		return marshaller(marshaller, resultTransformer, resultFactory, null);
 	}

-	public static Transformer marshaller(org.springframework.oxm.Marshaller marshaller,
-			org.springframework.integration.xml.transformer.ResultTransformer resultTransformer,
-			org.springframework.integration.xml.result.ResultFactory resultFactory,
-			boolean extractPayload) {
+	public static MarshallingTransformer marshaller(Marshaller marshaller, ResultTransformer resultTransformer,
+			ResultFactory resultFactory, boolean extractPayload) {
 		return marshaller(marshaller, resultTransformer, resultFactory, Boolean.valueOf(extractPayload));
 	}

-	private static Transformer marshaller(org.springframework.oxm.Marshaller marshaller,
-			org.springframework.integration.xml.transformer.ResultTransformer resultTransformer,
-			org.springframework.integration.xml.result.ResultFactory resultFactory,
-			Boolean extractPayload) {
+	private static MarshallingTransformer marshaller(Marshaller marshaller, ResultTransformer resultTransformer,
+			ResultFactory resultFactory, Boolean extractPayload) {
 		try {
-			org.springframework.integration.xml.transformer.MarshallingTransformer transformer =
-					new org.springframework.integration.xml.transformer.MarshallingTransformer(marshaller, resultTransformer);
+			MarshallingTransformer transformer = new MarshallingTransformer(marshaller, resultTransformer);
 			if (resultFactory != null) {
 				transformer.setResultFactory(resultFactory);
 			}
@@ -267,25 +269,22 @@ public abstract class Transformers {
 		}
 	}

-	public static Transformer unmarshaller(org.springframework.oxm.Unmarshaller unmarshaller) {
+	public static UnmarshallingTransformer unmarshaller(Unmarshaller unmarshaller) {
 		return unmarshaller(unmarshaller, null);
 	}

-	public static Transformer unmarshaller(org.springframework.oxm.Unmarshaller unmarshaller,
-			org.springframework.integration.xml.source.SourceFactory sourceFactory) {
+	public static UnmarshallingTransformer unmarshaller(Unmarshaller unmarshaller, SourceFactory sourceFactory) {
 		return unmarshaller(unmarshaller, sourceFactory, false);
 	}

-	public static Transformer unmarshaller(org.springframework.oxm.Unmarshaller unmarshaller,
+	public static UnmarshallingTransformer unmarshaller(Unmarshaller unmarshaller,
 			boolean alwaysUseSourceFactory) {
 		return unmarshaller(unmarshaller, null, alwaysUseSourceFactory);
 	}

-	public static Transformer unmarshaller(org.springframework.oxm.Unmarshaller unmarshaller,
-			org.springframework.integration.xml.source.SourceFactory sourceFactory,
+	public static UnmarshallingTransformer unmarshaller(Unmarshaller unmarshaller, SourceFactory sourceFactory,
 			boolean alwaysUseSourceFactory) {
-		org.springframework.integration.xml.transformer.UnmarshallingTransformer transformer =
-				new org.springframework.integration.xml.transformer.UnmarshallingTransformer(unmarshaller);
+		UnmarshallingTransformer transformer = new UnmarshallingTransformer(unmarshaller);
 		if(sourceFactory != null) {
 			transformer.setSourceFactory(sourceFactory);
 		}
@@ -294,37 +293,29 @@ public abstract class Transformers {
 		return transformer;
 	}

-	public static Transformer xmlSource() {
+	public static SourceCreatingTransformer xmlSource() {
 		return xmlSource(null);
 	}

-	public static Transformer xmlSource(org.springframework.integration.xml.source.SourceFactory sourceFactory) {
-		if (sourceFactory != null) {
-			return new org.springframework.integration.xml.transformer.SourceCreatingTransformer(sourceFactory);
-		}
-		else {
-			return new org.springframework.integration.xml.transformer.SourceCreatingTransformer();
-		}
+	public static SourceCreatingTransformer xmlSource(SourceFactory sourceFactory) {
+		return sourceFactory != null ? new SourceCreatingTransformer(sourceFactory) : new SourceCreatingTransformer();
 	}

-	public static Transformer xpath(String xpathExpression) {
+	public static XPathTransformer xpath(String xpathExpression) {
 		return xpath(xpathExpression, null, null);
 	}

-	public static Transformer xpath(String xpathExpression,
-			org.springframework.integration.xml.xpath.XPathEvaluationType xpathEvaluationType) {
+	public static XPathTransformer xpath(String xpathExpression, XPathEvaluationType xpathEvaluationType) {
 		return xpath(xpathExpression, xpathEvaluationType, null);
 	}

-	public static Transformer xpath(String xpathExpression,	org.springframework.xml.xpath.NodeMapper<?> nodeMapper) {
+	public static XPathTransformer xpath(String xpathExpression, NodeMapper<?> nodeMapper) {
 		return xpath(xpathExpression, null, nodeMapper);
 	}

-	public static Transformer xpath(String xpathExpression,
-			org.springframework.integration.xml.xpath.XPathEvaluationType xpathEvaluationType,
-			org.springframework.xml.xpath.NodeMapper<?> nodeMapper) {
-		org.springframework.integration.xml.transformer.XPathTransformer transformer =
-				new org.springframework.integration.xml.transformer.XPathTransformer(xpathExpression);
+	public static XPathTransformer xpath(String xpathExpression, XPathEvaluationType xpathEvaluationType,
+			NodeMapper<?> nodeMapper) {
+		XPathTransformer transformer = new XPathTransformer(xpathExpression);
 		if (xpathEvaluationType != null) {
 			transformer.setEvaluationType(xpathEvaluationType);
 		}
@@ -334,9 +325,8 @@ public abstract class Transformers {
 		return transformer;
 	}

-	public static Transformer xslt(Resource xsltTemplate, Tuple2<String, String>... xslParameterMappings) {
-		org.springframework.integration.xml.transformer.XsltPayloadTransformer transformer =
-				new org.springframework.integration.xml.transformer.XsltPayloadTransformer(xsltTemplate);
+	public static XsltPayloadTransformer xslt(Resource xsltTemplate, Tuple2<String, String>... xslParameterMappings) {
+		XsltPayloadTransformer transformer = new XsltPayloadTransformer(xsltTemplate);
 		if (xslParameterMappings != null) {
 			Map<String, Expression> params = new HashMap<String, Expression>(xslParameterMappings.length);
 			for (Tuple2<String, String> mapping : xslParameterMappings) {
--- a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java
+++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java
@@ -70,6 +70,7 @@ import org.springframework.data.mongodb.MongoDbFactory;
 import org.springframework.data.mongodb.core.SimpleMongoDbFactory;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.MessageDispatchingException;
+import org.springframework.integration.MessageRejectedException;
 import org.springframework.integration.annotation.Header;
 import org.springframework.integration.annotation.IntegrationComponentScan;
 import org.springframework.integration.annotation.MessageEndpoint;
@@ -121,6 +122,7 @@ import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.core.DestinationResolutionException;
 import org.springframework.messaging.support.ChannelInterceptorAdapter;
+import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.stereotype.Component;
 import org.springframework.test.annotation.DirtiesContext;
@@ -296,6 +298,13 @@ public class IntegrationFlowTests {
 	@Autowired
 	private Queue amqpQueue;

+	@Autowired
+	@Qualifier("gatewayInput")
+	private MessageChannel gatewayInput;
+
+	@Autowired
+	@Qualifier("gatewayError")
+	private PollableChannel gatewayError;

 	@BeforeClass
 	public static void setup() throws IOException {
@@ -852,7 +861,7 @@ public class IntegrationFlowTests {
 			assertNotNull(message);
 			assertEquals("hello " + i, message.getPayload());
 		}
-		assertNull(this.tailChannel.receive(10));
+		assertNull(this.tailChannel.receive(1));
 	}


@@ -862,6 +871,32 @@ public class IntegrationFlowTests {
 		assertEquals("HELLO WORLD", result);
 	}

+	@Test
+	public void testGatewayFlow() throws Exception {
+		PollableChannel replyChannel = new QueueChannel();
+		Message<String> message = MessageBuilder.withPayload("foo").setReplyChannel(replyChannel).build();
+
+		this.gatewayInput.send(message);
+
+		Message<?> receive = replyChannel.receive(2000);
+		assertNotNull(receive);
+		assertEquals("FOO", receive.getPayload());
+		assertNull(this.gatewayError.receive(1));
+
+		message = MessageBuilder.withPayload("bar").setReplyChannel(replyChannel).build();
+
+		this.gatewayInput.send(message);
+
+		receive = replyChannel.receive(1);
+		assertNull(receive);
+
+		receive = this.gatewayError.receive(2000);
+		assertNotNull(receive);
+		assertThat(receive, Matchers.instanceOf(ErrorMessage.class));
+		assertThat(receive.getPayload(), Matchers.instanceOf(MessageRejectedException.class));
+		assertThat(((Exception) receive.getPayload()).getMessage(), Matchers.containsString("' rejected Message"));
+	}
+
 	@MessagingGateway(defaultRequestChannel = "controlBus")
 	private static interface ControlBusGateway {

@@ -1292,6 +1327,27 @@ public class IntegrationFlowTests {
 					.transform((String p) -> p.toUpperCase())
 					.get();
 		}
+
+		@Bean
+		public IntegrationFlow gatewayFlow() {
+			return IntegrationFlows.from("gatewayInput")
+					.gateway("gatewayRequest", g -> g.errorChannel("gatewayError").replyTimeout(10L))
+					.get();
+		}
+
+		@Bean
+		public IntegrationFlow gatewayRequestFlow() {
+			return IntegrationFlows.from("gatewayRequest")
+					.filter("foo"::equals, f -> f.throwExceptionOnRejection(true))
+					.<String, String>transform(String::toUpperCase)
+					.get();
+		}
+
+		@Bean
+		public MessageChannel gatewayError() {
+			return MessageChannels.queue().get();
+		}
+
 	}

 	private static class RoutingTestBean {