From c3477801e6eae389c3b3065ce8e8b8b7074c2574 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 31 Oct 2014 19:11:49 +0200 Subject: [PATCH] DSL: `.gateway()` SubFlow support --- .../dsl/IntegrationFlowDefinition.java | 52 +++++++++---------- .../dsl/test/flows/IntegrationFlowTests.java | 3 +- 2 files changed, 28 insertions(+), 27 deletions(-) diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java index 76e9087..f1bfeb5 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java @@ -212,8 +212,7 @@ public abstract class IntegrationFlowDefinition B filter(GenericSelector

genericSelector, - Consumer endpointConfigurer) { + public

B filter(GenericSelector

genericSelector, Consumer endpointConfigurer) { return filter(null, genericSelector, endpointConfigurer); } @@ -227,13 +226,11 @@ public abstract class IntegrationFlowDefinition B handleWithAdapter( - Function> adapters) { + public B handleWithAdapter(Function> adapters) { return handleWithAdapter(adapters, null); } - public B handleWithAdapter( - Function> adapters, + public B handleWithAdapter(Function> adapters, Consumer> endpointConfigurer) { return handle(adapters.apply(new Adapters()), endpointConfigurer); } @@ -290,8 +287,7 @@ public abstract class IntegrationFlowDefinition B handle(H messageHandler, - Consumer> endpointConfigurer) { + public B handle(H messageHandler, Consumer> endpointConfigurer) { Assert.notNull(messageHandler); return this.register(new GenericEndpointSpec(messageHandler), endpointConfigurer); } @@ -529,8 +525,7 @@ public abstract class IntegrationFlowDefinition> routerConfigurer) { + public B route(String beanName, String method, Consumer> routerConfigurer) { return this.route(beanName, method, routerConfigurer, null); } @@ -548,8 +543,7 @@ public abstract class IntegrationFlowDefinition> routerConfigurer, + public B route(String expression, Consumer> routerConfigurer, Consumer> endpointConfigurer) { return this.route(new ExpressionEvaluatingRouter(PARSER.parseExpression(expression)), routerConfigurer, endpointConfigurer); @@ -559,8 +553,7 @@ public abstract class IntegrationFlowDefinition B route(Function router, - Consumer> routerConfigurer) { + public B route(Function router, Consumer> routerConfigurer) { return this.route(null, router, routerConfigurer); } @@ -573,8 +566,7 @@ public abstract class IntegrationFlowDefinition B route(Function router, - Consumer> routerConfigurer, + public B route(Function router, Consumer> routerConfigurer, Consumer> endpointConfigurer) { return route(null, router, routerConfigurer, endpointConfigurer); } @@ -588,8 +580,7 @@ public abstract class IntegrationFlowDefinition B route(R router, - Consumer> routerConfigurer, + public B route(R router, Consumer> routerConfigurer, Consumer> endpointConfigurer) { Collection componentsToRegister = null; if (routerConfigurer != null) { @@ -646,8 +637,7 @@ public abstract class IntegrationFlowDefinition B route(R router, - Consumer> endpointConfigurer) { + public B route(R router, Consumer> endpointConfigurer) { return handle(router, endpointConfigurer); } @@ -655,8 +645,7 @@ public abstract class IntegrationFlowDefinition endpointConfigurer) { + public B gateway(String requestChannel, Consumer endpointConfigurer) { return register(new GatewayEndpointSpec(requestChannel), endpointConfigurer); } @@ -664,13 +653,24 @@ public abstract class IntegrationFlowDefinition endpointConfigurer) { + public B gateway(MessageChannel requestChannel, Consumer endpointConfigurer) { return register(new GatewayEndpointSpec(requestChannel), endpointConfigurer); } - private > B register(S endpointSpec, - Consumer endpointConfigurer) { + public B gateway(IntegrationFlow flow) { + return gateway(flow, null); + } + + public B gateway(IntegrationFlow flow, Consumer endpointConfigurer) { + Assert.notNull(flow); + final DirectChannel requestChannel = new DirectChannel(); + IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(requestChannel); + flow.accept(flowBuilder); + addComponent(flowBuilder.get()); + return gateway(requestChannel, endpointConfigurer); + } + + private > B register(S endpointSpec, Consumer endpointConfigurer) { if (endpointConfigurer != null) { endpointConfigurer.accept(endpointSpec); } diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/flows/IntegrationFlowTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/flows/IntegrationFlowTests.java index 422be40..bf54125 100644 --- a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/flows/IntegrationFlowTests.java +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/flows/IntegrationFlowTests.java @@ -765,7 +765,7 @@ public class IntegrationFlowTests { Message receive = replyChannel.receive(2000); assertNotNull(receive); - assertEquals("FOO", receive.getPayload()); + assertEquals("From Gateway SubFlow: FOO", receive.getPayload()); assertNull(this.gatewayError.receive(1)); message = MessageBuilder.withPayload("bar").setReplyChannel(replyChannel).build(); @@ -1228,6 +1228,7 @@ public class IntegrationFlowTests { public IntegrationFlow gatewayFlow() { return IntegrationFlows.from("gatewayInput") .gateway("gatewayRequest", g -> g.errorChannel("gatewayError").replyTimeout(10L)) + .gateway(f -> f.transform("From Gateway SubFlow: "::concat)) .get(); }