diff --git a/spring-integration-java-dsl/build.gradle b/spring-integration-java-dsl/build.gradle index e97bbdd..de85b58 100644 --- a/spring-integration-java-dsl/build.gradle +++ b/spring-integration-java-dsl/build.gradle @@ -8,7 +8,6 @@ apply plugin: 'idea' group = 'org.springframework.integration' repositories { -// mavenLocal() if (version.endsWith('BUILD-SNAPSHOT')) { maven { url 'http://repo.spring.io/libs-snapshot' } } @@ -27,7 +26,7 @@ ext { embedMongoVersion = '1.45' jacocoVersion = '0.7.0.201403182114' log4jVersion = '1.2.17' - springIntegrationVersion = '4.0.1.RELEASE' + springIntegrationVersion = '4.0.2.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/IntegrationFlowBuilder.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java index 666e9b7..935c065 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 @@ -23,7 +23,6 @@ import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler; import org.springframework.integration.aggregator.AggregatingMessageHandler; import org.springframework.integration.aggregator.DefaultAggregatingMessageGroupProcessor; -import org.springframework.integration.aggregator.ResequencingMessageGroupProcessor; import org.springframework.integration.aggregator.ResequencingMessageHandler; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.FixedSubscriberChannel; @@ -64,7 +63,6 @@ import org.springframework.integration.transformer.ClaimCheckOutTransformer; import org.springframework.integration.transformer.ContentEnricher; import org.springframework.integration.transformer.ExpressionEvaluatingTransformer; import org.springframework.integration.transformer.GenericTransformer; -import org.springframework.integration.transformer.HeaderEnricher; import org.springframework.integration.transformer.HeaderFilter; import org.springframework.integration.transformer.MessageTransformingHandler; import org.springframework.integration.transformer.MethodInvokingTransformer; @@ -127,8 +125,11 @@ public final class IntegrationFlowBuilder { } public IntegrationFlowBuilder controlBus() { - return this.handle(new ServiceActivatingHandler(new ExpressionCommandMessageProcessor( - new ControlBusMethodFilter())), null); + return controlBus(null); + } + + public IntegrationFlowBuilder controlBus(EndpointConfigurer> endpointConfigurer) { + return this.handle(new ServiceActivatingHandler(new ExpressionCommandMessageProcessor(new ControlBusMethodFilter())), endpointConfigurer); } public IntegrationFlowBuilder transform(String expression) { @@ -259,16 +260,7 @@ public final class IntegrationFlowBuilder { Assert.notNull(enricherConfigurer); EnricherSpec enricherSpec = new EnricherSpec(); enricherConfigurer.configure(enricherSpec); - return this.enrich(enricherSpec.get(), endpointConfigurer); - } - - public IntegrationFlowBuilder enrich(ContentEnricher contentEnricher) { - return this.enrich(contentEnricher, null); - } - - public IntegrationFlowBuilder enrich(ContentEnricher contentEnricher, - EndpointConfigurer> endpointConfigurer) { - return this.handle(contentEnricher, endpointConfigurer); + return this.handle(enricherSpec.get(), endpointConfigurer); } public IntegrationFlowBuilder enrichHeaders(ComponentConfigurer headerEnricherConfigurer) { @@ -280,16 +272,7 @@ public final class IntegrationFlowBuilder { Assert.notNull(headerEnricherConfigurer); HeaderEnricherSpec headerEnricherSpec = new HeaderEnricherSpec(); headerEnricherConfigurer.configure(headerEnricherSpec); - return this.enrichHeaders(headerEnricherSpec.get(), endpointConfigurer); - } - - public IntegrationFlowBuilder enrichHeaders(HeaderEnricher headerEnricher) { - return this.enrichHeaders(headerEnricher, null); - } - - public IntegrationFlowBuilder enrichHeaders(HeaderEnricher headerEnricher, - EndpointConfigurer> endpointConfigurer) { - return this.addComponent(headerEnricher).transform(headerEnricher, endpointConfigurer); + return transform(headerEnricherSpec.get(), endpointConfigurer); } public IntegrationFlowBuilder split(EndpointConfigurer> endpointConfigurer) { @@ -393,12 +376,7 @@ public final class IntegrationFlowBuilder { } public IntegrationFlowBuilder resequence(EndpointConfigurer> endpointConfigurer) { - return this.resequence(new ResequencingMessageHandler(new ResequencingMessageGroupProcessor()), - endpointConfigurer); - } - - public IntegrationFlowBuilder resequence(ComponentConfigurer resequencerConfigurer) { - return this.resequence(resequencerConfigurer, null); + return this.handle(new ResequencerSpec().get(), endpointConfigurer); } public IntegrationFlowBuilder resequence(ComponentConfigurer resequencerConfigurer, @@ -406,20 +384,16 @@ public final class IntegrationFlowBuilder { Assert.notNull(resequencerConfigurer); ResequencerSpec spec = new ResequencerSpec(); resequencerConfigurer.configure(spec); - return this.resequence(spec.get(), endpointConfigurer); + return this.handle(spec.get(), endpointConfigurer); } - public IntegrationFlowBuilder resequence(ResequencingMessageHandler resequencer) { - return this.resequence(resequencer, null); + public IntegrationFlowBuilder aggregate() { + return aggregate((EndpointConfigurer>) null); } - public IntegrationFlowBuilder resequence(ResequencingMessageHandler resequencer, - EndpointConfigurer> endpointConfigurer) { - return this.handle(resequencer, endpointConfigurer); - } - - public IntegrationFlowBuilder aggregate(EndpointConfigurer> endpointConfigurer) { - return this.aggregate(new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor()), + public IntegrationFlowBuilder + aggregate(EndpointConfigurer> endpointConfigurer) { + return handle(new AggregatorSpec().outputProcessor(new DefaultAggregatingMessageGroupProcessor()).get(), endpointConfigurer); } @@ -428,15 +402,9 @@ public final class IntegrationFlowBuilder { Assert.notNull(aggregatorConfigurer); AggregatorSpec spec = new AggregatorSpec(); aggregatorConfigurer.configure(spec); - return this.aggregate(spec.get(), endpointConfigurer); + return this.handle(spec.get(), endpointConfigurer); } - public IntegrationFlowBuilder aggregate(AggregatingMessageHandler aggregator, - EndpointConfigurer> endpointConfigurer) { - return this.handle(aggregator, endpointConfigurer); - } - - public IntegrationFlowBuilder route(String beanName, String method) { return this.route(beanName, method, null); } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java index 250621a..47f791e 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java @@ -21,6 +21,7 @@ import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean; import org.springframework.integration.core.MessageSource; import org.springframework.integration.dsl.channel.MessageChannelSpec; +import org.springframework.integration.dsl.core.MessagingGatewaySpec; import org.springframework.integration.dsl.support.EndpointConfigurer; import org.springframework.integration.dsl.support.FixedSubscriberChannelPrototype; import org.springframework.integration.dsl.support.MessageChannelReference; @@ -88,6 +89,10 @@ public final class IntegrationFlows { return from(outputChannel).addComponent(messageProducer); } + public static IntegrationFlowBuilder from(MessagingGatewaySpec inboundGatewaySpec) { + return from(inboundGatewaySpec.get()); + } + public static IntegrationFlowBuilder from(MessagingGatewaySupport inboundGateway) { DirectFieldAccessor dfa = new DirectFieldAccessor(inboundGateway); MessageChannel outputChannel = (MessageChannel) dfa.getPropertyValue("requestChannel"); 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 ec6274b..17c3368 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 @@ -86,7 +86,6 @@ import org.springframework.integration.config.GlobalChannelInterceptor; import org.springframework.integration.core.MessageSource; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; -import org.springframework.integration.dsl.ResequencerSpec; import org.springframework.integration.dsl.amqp.Amqp; import org.springframework.integration.dsl.channel.DirectChannelSpec; import org.springframework.integration.dsl.channel.MessageChannels; @@ -1215,7 +1214,7 @@ public class IntegrationFlowTests { .channel(MessageChannels.executor(this.taskExecutor())) .transform(Integer::parseInt) .enrichHeaders(s -> s.headerExpression(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, "payload")) - .resequence((ResequencerSpec r) -> r.releasePartialSequences(true).correlationExpression("'foo'")) + .resequence(r -> r.releasePartialSequences(true).correlationExpression("'foo'"), null) .headerFilter("foo", false) .get(); } @@ -1226,7 +1225,7 @@ public class IntegrationFlowTests { .split(null) .channel(MessageChannels.executor(this.taskExecutor())) .resequence() - .aggregate(null) + .aggregate() .get(); } @@ -1316,7 +1315,7 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow amqpFlow() { - return IntegrationFlows.from(Amqp.inboundGateway(this.rabbitConnectionFactory, queue()).get()) + return IntegrationFlows.from(Amqp.inboundGateway(this.rabbitConnectionFactory, queue())) .transform("hello "::concat) .transform(String.class, String::toUpperCase) .get();