From 2a25a1960f405503fcff93f941337dd13eedf417 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 22 May 2014 20:43:38 +0300 Subject: [PATCH] DSL: add `gateway()` EIP-method * Upgrade dependencies * Polishing `Transformers` --- spring-integration-java-dsl/build.gradle | 2 +- .../integration/dsl/GatewayEndpointSpec.java | 67 ++++++++ .../dsl/GatewayMessageHandler.java | 116 +++++++++++++ .../dsl/IntegrationFlowBuilder.java | 82 +++++---- .../integration/dsl/amqp/Amqp.java | 1 - .../integration/dsl/support/Transformers.java | 160 ++++++++---------- .../dsl/test/IntegrationFlowTests.java | 58 ++++++- 7 files changed, 368 insertions(+), 118 deletions(-) create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GatewayEndpointSpec.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GatewayMessageHandler.java diff --git a/spring-integration-java-dsl/build.gradle b/spring-integration-java-dsl/build.gradle index d7b4a84..e97bbdd 100644 --- a/spring-integration-java-dsl/build.gradle +++ b/spring-integration-java-dsl/build.gradle @@ -27,7 +27,7 @@ ext { embedMongoVersion = '1.45' jacocoVersion = '0.7.0.201403182114' log4jVersion = '1.2.17' - springIntegrationVersion = '4.0.1.BUILD-SNAPSHOT' + springIntegrationVersion = '4.0.1.RELEASE' springBootVersion = '1.1.0.BUILD-SNAPSHOT' linkHomepage = 'https://github.com/spring-projects/spring-integration-extensions' diff --git 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 new file mode 100644 index 0000000..7c3a149 --- /dev/null +++ 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(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; + } + +} diff --git 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 new file mode 100644 index 0000000..4e6f0fe --- /dev/null +++ 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); + } + +} diff --git 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 index af9c370..362a2bc 100644 --- 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 IntegrationFlowBuilder transform(GenericTransformer genericTransformer, - EndpointConfigurer> endpointConfigurer) { + EndpointConfigurer> 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> endpointConfigurer) { + EndpointConfigurer> endpointConfigurer) { return this.handle(new ServiceActivatingHandler(new BeanNameMessageProcessor(beanName, methodName)), endpointConfigurer); } public IntegrationFlowBuilder handle(H messageHandler, - EndpointConfigurer> endpointConfigurer) { + EndpointConfigurer> endpointConfigurer) { Assert.notNull(messageHandler); return this.register(new GenericEndpointSpec(messageHandler), endpointConfigurer); } @@ -206,7 +206,7 @@ public final class IntegrationFlowBuilder { } public IntegrationFlowBuilder enrich(ComponentConfigurer enricherConfigurer, - EndpointConfigurer> endpointConfigurer) { + EndpointConfigurer> endpointConfigurer) { Assert.notNull(enricherConfigurer); EnricherSpec enricherSpec = new EnricherSpec(); enricherConfigurer.configure(enricherSpec); @@ -227,7 +227,7 @@ public final class IntegrationFlowBuilder { } public IntegrationFlowBuilder enrichHeaders(ComponentConfigurer headerEnricherConfigurer, - EndpointConfigurer> endpointConfigurer) { + EndpointConfigurer> 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> endpointConfigurer) { + EndpointConfigurer> endpointConfigurer) { return this.addComponent(headerEnricher).transform(headerEnricher, endpointConfigurer); } @@ -247,7 +247,8 @@ public final class IntegrationFlowBuilder { return this.split((EndpointConfigurer>) null); } - public IntegrationFlowBuilder split(EndpointConfigurer> endpointConfigurer) { + public + IntegrationFlowBuilder split(EndpointConfigurer> endpointConfigurer) { return this.split(new DefaultMessageSplitter(), endpointConfigurer); } @@ -265,7 +266,7 @@ public final class IntegrationFlowBuilder { } public IntegrationFlowBuilder split(String beanName, String methodName, - EndpointConfigurer> endpointConfigurer) { + EndpointConfigurer> endpointConfigurer) { return this.split(new MethodInvokingSplitter(new BeanNameMessageProcessor>(beanName, methodName)), endpointConfigurer); } @@ -279,7 +280,7 @@ public final class IntegrationFlowBuilder { } public IntegrationFlowBuilder split(GenericSplitter splitter, - EndpointConfigurer> endpointConfigurer) { + EndpointConfigurer> 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> endpointConfigurer) { + EndpointConfigurer> endpointConfigurer) { return this.transform(headerFilter, endpointConfigurer); } @@ -336,7 +338,7 @@ public final class IntegrationFlowBuilder { } public IntegrationFlowBuilder claimCheckOut(MessageStore messageStore, boolean removeMessage, - EndpointConfigurer> endpointConfigurer) { + EndpointConfigurer> 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>) null); } - public IntegrationFlowBuilder resequence(EndpointConfigurer> endpointConfigurer) { - return this.resequence(new ResequencingMessageHandler(new ResequencingMessageGroupProcessor()), endpointConfigurer); + public + IntegrationFlowBuilder resequence(EndpointConfigurer> endpointConfigurer) { + return this.resequence(new ResequencingMessageHandler(new ResequencingMessageGroupProcessor()), + endpointConfigurer); } public IntegrationFlowBuilder resequence(ComponentConfigurer resequencerConfigurer) { @@ -355,7 +359,7 @@ public final class IntegrationFlowBuilder { } public IntegrationFlowBuilder resequence(ComponentConfigurer resequencerConfigurer, - EndpointConfigurer> endpointConfigurer) { + EndpointConfigurer> 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> endpointConfigurer) { + EndpointConfigurer> endpointConfigurer) { return this.handle(resequencer, endpointConfigurer); } @@ -375,7 +379,8 @@ public final class IntegrationFlowBuilder { return this.aggregate((EndpointConfigurer>) null); } - public IntegrationFlowBuilder aggregate(EndpointConfigurer> endpointConfigurer) { + public + IntegrationFlowBuilder aggregate(EndpointConfigurer> endpointConfigurer) { return this.aggregate(new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor()), endpointConfigurer); } @@ -385,7 +390,7 @@ public final class IntegrationFlowBuilder { } public IntegrationFlowBuilder aggregate(ComponentConfigurer aggregatorConfigurer, - EndpointConfigurer> endpointConfigurer) { + EndpointConfigurer> 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> endpointConfigurer) { + EndpointConfigurer> endpointConfigurer) { return this.handle(aggregator, endpointConfigurer); } @@ -407,13 +412,13 @@ public final class IntegrationFlowBuilder { } public IntegrationFlowBuilder route(String beanName, String method, - ComponentConfigurer> routerConfigurer) { + ComponentConfigurer> routerConfigurer) { return this.route(beanName, method, routerConfigurer, null); } public IntegrationFlowBuilder route(String beanName, String method, - ComponentConfigurer> routerConfigurer, - EndpointConfigurer> endpointConfigurer) { + ComponentConfigurer> routerConfigurer, + EndpointConfigurer> endpointConfigurer) { return this.route(new MethodInvokingRouter(new BeanNameMessageProcessor(beanName, method)), routerConfigurer, endpointConfigurer); } @@ -429,8 +434,8 @@ public final class IntegrationFlowBuilder { } public IntegrationFlowBuilder route(String expression, - ComponentConfigurer> routerConfigurer, - EndpointConfigurer> endpointConfigurer) { + ComponentConfigurer> routerConfigurer, + EndpointConfigurer> endpointConfigurer) { return this.route(new ExpressionEvaluatingRouter(PARSER.parseExpression(expression)), routerConfigurer, endpointConfigurer); } @@ -445,8 +450,8 @@ public final class IntegrationFlowBuilder { } public IntegrationFlowBuilder route(GenericRouter router, - ComponentConfigurer> routerConfigurer, - EndpointConfigurer> endpointConfigurer) { + ComponentConfigurer> routerConfigurer, + EndpointConfigurer> endpointConfigurer) { return this.route(new MethodInvokingRouter(router), routerConfigurer, endpointConfigurer); } @@ -456,8 +461,8 @@ public final class IntegrationFlowBuilder { } public IntegrationFlowBuilder route(R router, - ComponentConfigurer> routerConfigurer, - EndpointConfigurer> endpointConfigurer) { + ComponentConfigurer> routerConfigurer, + EndpointConfigurer> endpointConfigurer) { if (routerConfigurer != null) { RouterSpec routerSpec = new RouterSpec(router); routerConfigurer.configure(routerSpec); @@ -470,7 +475,7 @@ public final class IntegrationFlowBuilder { } public IntegrationFlowBuilder recipientListRoute(ComponentConfigurer routerConfigurer, - EndpointConfigurer> endpointConfigurer) { + EndpointConfigurer> endpointConfigurer) { Assert.notNull(routerConfigurer); RecipientListRouterSpec spec = new RecipientListRouterSpec(); routerConfigurer.configure(spec); @@ -484,10 +489,27 @@ public final class IntegrationFlowBuilder { } public IntegrationFlowBuilder route(R router, - EndpointConfigurer> endpointConfigurer) { + EndpointConfigurer> endpointConfigurer) { return this.handle(router, endpointConfigurer); } + public IntegrationFlowBuilder gateway(String requestChannel) { + return gateway(requestChannel, null); + } + + public IntegrationFlowBuilder gateway(String requestChannel, + EndpointConfigurer endpointConfigurer) { + return register(new GatewayEndpointSpec(requestChannel), endpointConfigurer); + } + + public IntegrationFlowBuilder gateway(MessageChannel requestChannel) { + return gateway(requestChannel, null); + } + + public IntegrationFlowBuilder gateway(MessageChannel requestChannel, + EndpointConfigurer endpointConfigurer) { + return register(new GatewayEndpointSpec(requestChannel), endpointConfigurer); + } private > IntegrationFlowBuilder register(S endpointSpec, EndpointConfigurer endpointConfigurer) { diff --git 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 index 16120b6..cf4e342 100644 --- 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 { diff --git 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 index abf4b1b..993c98d 100644 --- 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 serializer) { + public static PayloadSerializingTransformer serializer(Serializer 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 deserializer) { + public static PayloadDeserializingTransformer deserializer(Deserializer deserializer) { PayloadDeserializingTransformer transformer = new PayloadDeserializingTransformer(); if (deserializer != null) { transformer.setDeserializer(deserializer); @@ -162,98 +176,86 @@ public abstract class Transformers { return transformer; } - public static Transformer converter(Converter converter) { + public static PayloadTypeConvertingTransformer converter(Converter converter) { Assert.notNull(converter, "The Converter is required for the PayloadTypeConvertingTransformer"); PayloadTypeConvertingTransformer transformer = new PayloadTypeConvertingTransformer(); 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... xslParameterMappings) { - org.springframework.integration.xml.transformer.XsltPayloadTransformer transformer = - new org.springframework.integration.xml.transformer.XsltPayloadTransformer(xsltTemplate); + public static XsltPayloadTransformer xslt(Resource xsltTemplate, Tuple2... xslParameterMappings) { + XsltPayloadTransformer transformer = new XsltPayloadTransformer(xsltTemplate); if (xslParameterMappings != null) { Map params = new HashMap(xslParameterMappings.length); for (Tuple2 mapping : xslParameterMappings) { diff --git 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 index e5e4de2..c82fe01 100644 --- 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 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)) + .transform(String::toUpperCase) + .get(); + } + + @Bean + public MessageChannel gatewayError() { + return MessageChannels.queue().get(); + } + } private static class RoutingTestBean {