Spring/spring-integration-extensions
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DSL: Clean up some IntegrationFlowBuilder API

Browse Source
This commit is contained in:
Artem Bilan
2014-06-02 16:57:33 +03:00
parent f7529453b4
commit 2dd27c5df9
4 changed files with 24 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java
View File

@@ -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<GenericEndpointSpec<ServiceActivatingHandler>> 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<GenericEndpointSpec<ContentEnricher>> endpointConfigurer) {
return this.handle(contentEnricher, endpointConfigurer);
return this.handle(enricherSpec.get(), endpointConfigurer);
}
public IntegrationFlowBuilder enrichHeaders(ComponentConfigurer<HeaderEnricherSpec> 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<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
return this.addComponent(headerEnricher).transform(headerEnricher, endpointConfigurer);
return transform(headerEnricherSpec.get(), endpointConfigurer);
}
public IntegrationFlowBuilder split(EndpointConfigurer<SplitterEndpointSpec<DefaultMessageSplitter>> endpointConfigurer) {
@@ -393,12 +376,7 @@ public final class IntegrationFlowBuilder {
}
public IntegrationFlowBuilder resequence(EndpointConfigurer<GenericEndpointSpec<ResequencingMessageHandler>> endpointConfigurer) {
return this.resequence(new ResequencingMessageHandler(new ResequencingMessageGroupProcessor()),
endpointConfigurer);
}
public IntegrationFlowBuilder resequence(ComponentConfigurer<ResequencerSpec> resequencerConfigurer) {
return this.resequence(resequencerConfigurer, null);
return this.handle(new ResequencerSpec().get(), endpointConfigurer);
}
public IntegrationFlowBuilder resequence(ComponentConfigurer<ResequencerSpec> 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<GenericEndpointSpec<AggregatingMessageHandler>>) null);
}
public IntegrationFlowBuilder resequence(ResequencingMessageHandler resequencer,
EndpointConfigurer<GenericEndpointSpec<ResequencingMessageHandler>> endpointConfigurer) {
return this.handle(resequencer, endpointConfigurer);
}
public IntegrationFlowBuilder aggregate(EndpointConfigurer<GenericEndpointSpec<AggregatingMessageHandler>> endpointConfigurer) {
return this.aggregate(new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor()),
public IntegrationFlowBuilder
aggregate(EndpointConfigurer<GenericEndpointSpec<AggregatingMessageHandler>> 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<GenericEndpointSpec<AggregatingMessageHandler>> endpointConfigurer) {
return this.handle(aggregator, endpointConfigurer);
}
public IntegrationFlowBuilder route(String beanName, String method) {
return this.route(beanName, method, null);
}

5
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java
View File

@@ -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");

7
spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java
View File

@@ -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()))
.<String, Integer>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();