diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java index 4df1cdab42..386ae7a54e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java @@ -29,7 +29,9 @@ import org.springframework.messaging.Message; * * @since 5.0 */ -public class Channels { +public final class Channels { + + static final Channels INSTANCE = new Channels(); public DirectChannelSpec direct() { return MessageChannels.direct(); @@ -129,7 +131,7 @@ public class Channels { return MessageChannels.flux(id); } - Channels() { + private Channels() { super(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java index 65ad077732..1629a49e6f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java @@ -16,6 +16,10 @@ package org.springframework.integration.dsl; +import org.reactivestreams.Publisher; + +import org.springframework.messaging.Message; + /** * @author Artem Bilan * @@ -32,4 +36,9 @@ public final class IntegrationFlowBuilder extends IntegrationFlowDefinition Publisher> toReactivePublisher() { // NOSONAR - not useless, increases visibility + return super.toReactivePublisher(); + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java index e5fd8ed586..ff0777692c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java @@ -231,7 +231,7 @@ public abstract class IntegrationFlowDefinition> channels) { Assert.notNull(channels, "'channels' must not be null"); - return channel(channels.apply(new Channels())); + return channel(channels.apply(Channels.INSTANCE)); } /** @@ -529,6 +529,7 @@ public abstract class IntegrationFlowDefinition> endpointConfigurer) { + MethodInvokingTransformer transformer; if (StringUtils.hasText(methodName)) { transformer = new MethodInvokingTransformer(service, methodName); @@ -588,6 +589,7 @@ public abstract class IntegrationFlowDefinition messageProcessorSpec, Consumer> endpointConfigurer) { + Assert.notNull(messageProcessorSpec, MESSAGE_PROCESSOR_SPEC_MUST_NOT_BE_NULL); MessageProcessor processor = messageProcessorSpec.get(); return addComponent(processor) @@ -614,8 +616,6 @@ public abstract class IntegrationFlowDefinition B transform(Class

payloadType, GenericTransformer genericTransformer, Consumer> endpointConfigurer) { + Assert.notNull(genericTransformer, "'genericTransformer' must not be null"); Transformer transformer = genericTransformer instanceof Transformer ? (Transformer) genericTransformer : (ClassUtils.isLambda(genericTransformer.getClass()) @@ -759,13 +760,10 @@ public abstract class IntegrationFlowDefinition endpointConfigurer) { - MethodInvokingSelector selector; - if (StringUtils.hasText(methodName)) { - selector = new MethodInvokingSelector(service, methodName); - } - else { - selector = new MethodInvokingSelector(service); - } + MethodInvokingSelector selector = + StringUtils.hasText(methodName) + ? new MethodInvokingSelector(service, methodName) + : new MethodInvokingSelector(service); return filter(selector, endpointConfigurer); } @@ -891,6 +889,7 @@ public abstract class IntegrationFlowDefinition B filter(Class

payloadType, GenericSelector

genericSelector, Consumer endpointConfigurer) { + Assert.notNull(genericSelector, "'genericSelector' must not be null"); MessageSelector selector = genericSelector instanceof MessageSelector ? (MessageSelector) genericSelector : (ClassUtils.isLambda(genericSelector.getClass()) @@ -957,7 +956,7 @@ public abstract class IntegrationFlowDefinition> endpointConfigurer) { - return handle(new ServiceActivatingHandler(new BeanNameMessageProcessor(beanName, methodName)), + return handle(new ServiceActivatingHandler(new BeanNameMessageProcessor<>(beanName, methodName)), endpointConfigurer); } @@ -997,6 +996,7 @@ public abstract class IntegrationFlowDefinition> endpointConfigurer) { + ServiceActivatingHandler handler; if (StringUtils.hasText(methodName)) { handler = new ServiceActivatingHandler(service, methodName); @@ -1050,6 +1050,7 @@ public abstract class IntegrationFlowDefinition B handle(GenericHandler

handler, Consumer> endpointConfigurer) { + return handle(null, handler, endpointConfigurer); } @@ -1097,6 +1098,7 @@ public abstract class IntegrationFlowDefinition B handle(Class

payloadType, GenericHandler

handler, Consumer> endpointConfigurer) { + ServiceActivatingHandler serviceActivatingHandler; if (ClassUtils.isLambda(handler.getClass())) { serviceActivatingHandler = new ServiceActivatingHandler(new LambdaMessageProcessor(handler, payloadType)); @@ -1139,6 +1141,7 @@ public abstract class IntegrationFlowDefinition messageProcessorSpec, Consumer> endpointConfigurer) { + Assert.notNull(messageProcessorSpec, MESSAGE_PROCESSOR_SPEC_MUST_NOT_BE_NULL); MessageProcessor processor = messageProcessorSpec.get(); return addComponent(processor) @@ -1164,6 +1167,7 @@ public abstract class IntegrationFlowDefinition B handle(MessageHandlerSpec messageHandlerSpec, Consumer> endpointConfigurer) { + Assert.notNull(messageHandlerSpec, "'messageHandlerSpec' must not be null"); if (messageHandlerSpec instanceof ComponentsRegistration) { addComponents(((ComponentsRegistration) messageHandlerSpec).getComponentsToRegister()); @@ -1188,7 +1192,7 @@ public abstract class IntegrationFlowDefinition B handle(H messageHandler, Consumer> endpointConfigurer) { Assert.notNull(messageHandler, "'messageHandler' must not be null"); - return this.register(new GenericEndpointSpec<>(messageHandler), endpointConfigurer); + return register(new GenericEndpointSpec<>(messageHandler), endpointConfigurer); } /** @@ -1306,6 +1310,7 @@ public abstract class IntegrationFlowDefinition headers, Consumer> endpointConfigurer) { + return enrichHeaders(headers.get(), endpointConfigurer); } @@ -1331,8 +1336,9 @@ public abstract class IntegrationFlowDefinition headers, + public B enrichHeaders(Map headers, Consumer> endpointConfigurer) { + HeaderEnricherSpec headerEnricherSpec = new HeaderEnricherSpec(); headerEnricherSpec.headers(headers); Tuple2 tuple2 = headerEnricherSpec.get(); @@ -1450,6 +1456,7 @@ public abstract class IntegrationFlowDefinition> endpointConfigurer) { + MethodInvokingSplitter splitter; if (StringUtils.hasText(methodName)) { splitter = new MethodInvokingSplitter(service, methodName); @@ -1484,7 +1491,8 @@ public abstract class IntegrationFlowDefinition> endpointConfigurer) { - return split(new MethodInvokingSplitter(new BeanNameMessageProcessor(beanName, methodName)), + + return split(new MethodInvokingSplitter(new BeanNameMessageProcessor<>(beanName, methodName)), endpointConfigurer); } @@ -1524,6 +1532,7 @@ public abstract class IntegrationFlowDefinition messageProcessorSpec, Consumer> endpointConfigurer) { + Assert.notNull(messageProcessorSpec, MESSAGE_PROCESSOR_SPEC_MUST_NOT_BE_NULL); MessageProcessor processor = messageProcessorSpec.get(); return addComponent(processor) @@ -1581,6 +1590,7 @@ public abstract class IntegrationFlowDefinition B split(Function splitter, Consumer> endpointConfigurer) { + return split(null, splitter, endpointConfigurer); } @@ -1669,8 +1679,9 @@ public abstract class IntegrationFlowDefinition B split(S splitter, Consumer> endpointConfigurer) { + Assert.notNull(splitter, "'splitter' must not be null"); - return this.register(new SplitterEndpointSpec<>(splitter), endpointConfigurer); + return register(new SplitterEndpointSpec<>(splitter), endpointConfigurer); } /** @@ -1680,7 +1691,7 @@ public abstract class IntegrationFlowDefinition> endpointConfigurer) { + return transform(headerFilter, endpointConfigurer); } @@ -1731,6 +1743,7 @@ public abstract class IntegrationFlowDefinition> endpointConfigurer) { + return transform(new ClaimCheckInTransformer(messageStore), endpointConfigurer); } @@ -1742,7 +1755,7 @@ public abstract class IntegrationFlowDefinition> endpointConfigurer) { + ClaimCheckOutTransformer claimCheckOutTransformer = new ClaimCheckOutTransformer(messageStore); claimCheckOutTransformer.setRemoveMessage(removeMessage); return transform(claimCheckOutTransformer, endpointConfigurer); @@ -1855,6 +1869,7 @@ public abstract class IntegrationFlowDefinition> routerConfigurer) { + MethodInvokingRouter methodInvokingRouter = new MethodInvokingRouter(new BeanNameMessageProcessor<>(beanName, method)); return route(new RouterSpec<>(methodInvokingRouter), routerConfigurer); @@ -1894,6 +1909,7 @@ public abstract class IntegrationFlowDefinition> routerConfigurer) { + MethodInvokingRouter router; if (StringUtils.hasText(methodName)) { router = new MethodInvokingRouter(service, methodName); @@ -2062,6 +2078,7 @@ public abstract class IntegrationFlowDefinition messageProcessorSpec, Consumer> routerConfigurer) { + Assert.notNull(messageProcessorSpec, MESSAGE_PROCESSOR_SPEC_MUST_NOT_BE_NULL); MessageProcessor processor = messageProcessorSpec.get(); addComponent(processor); @@ -2272,7 +2289,6 @@ public abstract class IntegrationFlowDefinition endpointConfigurer) { MessageChannel requestChannel = obtainInputChannelFromFlow(flow); - return gateway(requestChannel, endpointConfigurer); } @@ -2826,11 +2842,11 @@ public abstract class IntegrationFlowDefinition gatherer, Consumer scatterGather) { + AggregatorSpec aggregatorSpec = new AggregatorSpec(); if (gatherer != null) { gatherer.accept(aggregatorSpec); } - AggregatingMessageHandler aggregatingMessageHandler = aggregatorSpec.get().getT2(); addComponent(aggregatingMessageHandler); ScatterGatherHandler messageHandler = new ScatterGatherHandler(scatterChannel, aggregatingMessageHandler); @@ -2882,7 +2898,6 @@ public abstract class IntegrationFlowDefinition> endpointConfigurer) { + MessageProcessor trigger = new BeanNameMessageProcessor<>(triggerActionId, "trigger"); return handle(new ServiceActivatingHandler(trigger), endpointConfigurer); } @@ -2955,6 +2971,7 @@ public abstract class IntegrationFlowDefinition> endpointConfigurer) { + return handle(new ServiceActivatingHandler(triggerAction, "trigger"), endpointConfigurer); } @@ -2991,7 +3008,7 @@ public abstract class IntegrationFlowDefinition Publisher> toReactivePublisher() { + protected Publisher> toReactivePublisher() { MessageChannel channelForPublisher = this.currentMessageChannel; Publisher> publisher; if (channelForPublisher instanceof Publisher) {