diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java index a633fccef6..01003c9d67 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java @@ -668,9 +668,9 @@ public abstract class BaseIntegrationFlowDefinition B transform(Class

payloadType, GenericTransformer genericTransformer) { - return transform(payloadType, genericTransformer, null); + public B transform(Class

expectedType, GenericTransformer genericTransformer) { + return transform(expectedType, genericTransformer, null); } /** * Populate the {@link MessageTransformingHandler} instance * for the provided {@code payloadType} to convert at runtime. - * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. + * In addition, accept options for the integration endpoint using {@link GenericEndpointSpec}. * @param payloadType the {@link Class} for expected payload type. * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. * @param

the payload type - 'transform to'. @@ -706,9 +706,9 @@ public abstract class BaseIntegrationFlowDefinition B transform(Class

payloadType, GenericTransformer genericTransformer, + public B transform(Class

expectedType, GenericTransformer genericTransformer, Consumer> endpointConfigurer) { Assert.notNull(genericTransformer, "'genericTransformer' must not be null"); Transformer transformer = genericTransformer instanceof Transformer ? (Transformer) genericTransformer : (ClassUtils.isLambda(genericTransformer.getClass()) - ? new MethodInvokingTransformer(new LambdaMessageProcessor(genericTransformer, payloadType)) + ? new MethodInvokingTransformer(new LambdaMessageProcessor(genericTransformer, expectedType)) : new MethodInvokingTransformer(genericTransformer, ClassUtils.TRANSFORMER_TRANSFORM_METHOD)); return addComponent(transformer) .handle(new MessageTransformingHandler(transformer), endpointConfigurer); @@ -840,13 +840,13 @@ public abstract class BaseIntegrationFlowDefinition * {@code * .filter(Date.class, p -> p.after(new Date())) * } * - * @param payloadType the {@link Class} for expected payload type. It can also be + * @param expectedType the {@link Class} for expected payload type. It can also be * {@code Message.class} if you wish to access the entire message in the selector. * Conversion to this type will be attempted, if necessary. * @param genericSelector the {@link GenericSelector} to use. @@ -854,21 +854,21 @@ public abstract class BaseIntegrationFlowDefinition B filter(Class

payloadType, GenericSelector

genericSelector) { - return filter(payloadType, genericSelector, null); + public

B filter(Class

expectedType, GenericSelector

genericSelector) { + return filter(expectedType, genericSelector, null); } /** * Populate a {@link MessageFilter} with {@link MethodInvokingSelector} * for the provided {@link GenericSelector}. - * In addition accept options for the integration endpoint using {@link FilterEndpointSpec}. - * Typically used with a Java 8 Lambda expression: + * In addition, accept options for the integration endpoint using {@link FilterEndpointSpec}. + * Typically, used with a Java 8 Lambda expression: *

 	 * {@code
 	 *  .filter(Date.class, p -> p.after(new Date()), e -> e.autoStartup(false))
 	 * }
 	 * 
- * @param payloadType the {@link Class} for expected payload type. It can also be + * @param expectedType the {@link Class} for expected payload type. It can also be * {@code Message.class} if you wish to access the entire message in the selector. * Conversion to this type will be attempted, if necessary. * @param genericSelector the {@link GenericSelector} to use. @@ -878,13 +878,13 @@ public abstract class BaseIntegrationFlowDefinition B filter(Class

payloadType, GenericSelector

genericSelector, + public

B filter(Class

expectedType, GenericSelector

genericSelector, Consumer endpointConfigurer) { Assert.notNull(genericSelector, "'genericSelector' must not be null"); MessageSelector selector = genericSelector instanceof MessageSelector ? (MessageSelector) genericSelector : (ClassUtils.isLambda(genericSelector.getClass()) - ? new MethodInvokingSelector(new LambdaMessageProcessor(genericSelector, payloadType)) + ? new MethodInvokingSelector(new LambdaMessageProcessor(genericSelector, expectedType)) : new MethodInvokingSelector(genericSelector, ClassUtils.SELECTOR_ACCEPT_METHOD)); return this.register(new FilterEndpointSpec(new MessageFilter(selector)), endpointConfigurer); } @@ -1001,13 +1001,13 @@ public abstract class BaseIntegrationFlowDefinition * {@code * .handle(Integer.class, (p, h) -> p / 2) * } * - * @param payloadType the {@link Class} for expected payload type. It can also be + * @param expectedType the {@link Class} for expected payload type. It can also be * {@code Message.class} if you wish to access the entire message in the handler. * Conversion to this type will be attempted, if necessary. * @param handler the handler to invoke. @@ -1015,8 +1015,8 @@ public abstract class BaseIntegrationFlowDefinition B handle(Class

payloadType, GenericHandler

handler) { - return handle(payloadType, handler, null); + public

B handle(Class

expectedType, GenericHandler

handler) { + return handle(expectedType, handler, null); } /** @@ -1030,7 +1030,7 @@ public abstract class BaseIntegrationFlowDefinition p / 2, e -> e.autoStartup(false)) * } * - * @param payloadType the {@link Class} for expected payload type. It can also be + * @param expectedType the {@link Class} for expected payload type. It can also be * {@code Message.class} if you wish to access the entire message in the handler. * Conversion to this type will be attempted, if necessary. * @param handler the handler to invoke. @@ -1039,15 +1039,15 @@ public abstract class BaseIntegrationFlowDefinition B handle(Class

payloadType, GenericHandler

handler, + public

B handle(Class

expectedType, GenericHandler

handler, Consumer> endpointConfigurer) { ServiceActivatingHandler serviceActivatingHandler; if (ClassUtils.isLambda(handler.getClass())) { - serviceActivatingHandler = new ServiceActivatingHandler(new LambdaMessageProcessor(handler, payloadType)); + serviceActivatingHandler = new ServiceActivatingHandler(new LambdaMessageProcessor(handler, expectedType)); } - else if (payloadType != null) { - return handle(payloadType, handler::handle, endpointConfigurer); + else if (expectedType != null) { + return handle(expectedType, handler::handle, endpointConfigurer); } else { serviceActivatingHandler = new ServiceActivatingHandler(handler, ClassUtils.HANDLER_HANDLE_METHOD); @@ -1488,7 +1488,7 @@ public abstract class BaseIntegrationFlowDefinition * {@code * .split(String.class, p -> @@ -1499,7 +1499,7 @@ public abstract class BaseIntegrationFlowDefinition - * @param payloadType the {@link Class} for expected payload type. It can also be + * @param expectedType the {@link Class} for expected payload type. It can also be * {@code Message.class} if you wish to access the entire message in the splitter. * Conversion to this type will be attempted, if necessary. * @param splitter the splitter {@link Function}. @@ -1507,15 +1507,15 @@ public abstract class BaseIntegrationFlowDefinition B split(Class

payloadType, Function splitter) { - return split(payloadType, splitter, null); + public

B split(Class

expectedType, Function splitter) { + return split(expectedType, splitter, null); } /** * Populate the {@link MethodInvokingSplitter} to evaluate the provided * {@link Function} at runtime. - * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. - * Typically used with a Java 8 Lambda expression: + * In addition, accept options for the integration endpoint using {@link GenericEndpointSpec}. + * Typically, used with a Java 8 Lambda expression: *

 	 * {@code
 	 *  .split(String.class, p ->
@@ -1527,7 +1527,7 @@ public abstract class BaseIntegrationFlowDefinition e.applySequence(false))
 	 * }
 	 * 
- * @param payloadType the {@link Class} for expected payload type. It can also be + * @param expectedType the {@link Class} for expected payload type. It can also be * {@code Message.class} if you wish to access the entire message in the splitter. * Conversion to this type will be attempted, if necessary. * @param splitter the splitter {@link Function}. @@ -1537,12 +1537,12 @@ public abstract class BaseIntegrationFlowDefinition B split(Class

payloadType, Function splitter, + public

B split(Class

expectedType, Function splitter, Consumer> endpointConfigurer) { MethodInvokingSplitter split = ClassUtils.isLambda(splitter.getClass()) - ? new MethodInvokingSplitter(new LambdaMessageProcessor(splitter, payloadType)) + ? new MethodInvokingSplitter(new LambdaMessageProcessor(splitter, expectedType)) : new MethodInvokingSplitter(splitter, ClassUtils.FUNCTION_APPLY_METHOD); return split(split, endpointConfigurer); } @@ -1880,8 +1880,8 @@ public abstract class BaseIntegrationFlowDefinition p % 2 == 0) * } * - * @param payloadType the {@link Class} for expected payload type. It can also be - * {@code Message.class} if you wish to access the entire message in the splitter. + * @param expectedType the {@link Class} for expected payload type. It can also be + * {@code Message.class} if you wish to access the entire message in the router. * Conversion to this type will be attempted, if necessary. * @param router the {@link Function} to use. * @param the source payload type or {@code Message.class}. @@ -1889,15 +1889,15 @@ public abstract class BaseIntegrationFlowDefinition B route(Class payloadType, Function router) { - return route(payloadType, router, null); + public B route(Class expectedType, Function router) { + return route(expectedType, router, null); } /** * Populate the {@link MethodInvokingRouter} for provided {@link Function} * and payload type and options from {@link RouterSpec}. - * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. - * Typically used with a Java 8 Lambda expression: + * In addition, accept options for the integration endpoint using {@link GenericEndpointSpec}. + * Typically, used with a Java 8 Lambda expression: *

 	 * {@code
 	 *  .route(Integer.class, p -> p % 2 == 0,
@@ -1907,8 +1907,8 @@ public abstract class BaseIntegrationFlowDefinition
-	 * @param payloadType the {@link Class} for expected payload type. It can also be
-	 * {@code Message.class} if you wish to access the entire message in the splitter.
+	 * @param expectedType the {@link Class} for expected payload type. It can also be
+	 * {@code Message.class} if you wish to access the entire message in the router.
 	 * Conversion to this type will be attempted, if necessary.
 	 * @param router the {@link Function} to use.
 	 * @param routerConfigurer the {@link Consumer} to provide {@link MethodInvokingRouter} options.
@@ -1917,12 +1917,12 @@ public abstract class BaseIntegrationFlowDefinition B route(Class

payloadType, Function router, + public B route(Class

expectedType, Function router, Consumer> routerConfigurer) { MethodInvokingRouter methodInvokingRouter = ClassUtils.isLambda(router.getClass()) - ? new MethodInvokingRouter(new LambdaMessageProcessor(router, payloadType)) + ? new MethodInvokingRouter(new LambdaMessageProcessor(router, expectedType)) : new MethodInvokingRouter(router, ClassUtils.FUNCTION_APPLY_METHOD); return route(new RouterSpec<>(methodInvokingRouter), routerConfigurer); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java index 2ed45478b2..063f0d2941 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java @@ -30,6 +30,7 @@ import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.core.MethodIntrospector; import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.converter.MessageConverter; import org.springframework.util.Assert; @@ -53,13 +54,19 @@ public class LambdaMessageProcessor implements MessageProcessor, BeanFac private final Method method; - private final Class payloadType; + @Nullable + private final Class expectedType; private final Class[] parameterTypes; private MessageConverter messageConverter; - public LambdaMessageProcessor(Object target, Class payloadType) { + /** + * Create a processor to evaluate a provided lambda at runtime against a request message. + * @param target the lambda object. + * @param expectedType an optional expected type for method argument conversion. + */ + public LambdaMessageProcessor(Object target, @Nullable Class expectedType) { Assert.notNull(target, "'target' must not be null"); this.target = target; @@ -79,7 +86,7 @@ public class LambdaMessageProcessor implements MessageProcessor, BeanFac this.method = methods.iterator().next(); this.method.setAccessible(true); this.parameterTypes = this.method.getParameterTypes(); - this.payloadType = payloadType; + this.expectedType = expectedType; } @Override @@ -138,13 +145,13 @@ public class LambdaMessageProcessor implements MessageProcessor, BeanFac } } else { - if (this.payloadType != null && - !ClassUtils.isAssignable(this.payloadType, message.getPayload().getClass())) { - if (Message.class.isAssignableFrom(this.payloadType)) { + if (this.expectedType != null && + !ClassUtils.isAssignable(this.expectedType, message.getPayload().getClass())) { + if (Message.class.isAssignableFrom(this.expectedType)) { args[i] = message; } else { - args[i] = this.messageConverter.fromMessage(message, this.payloadType); + args[i] = this.messageConverter.fromMessage(message, this.expectedType); } } else { diff --git a/src/reference/asciidoc/router.adoc b/src/reference/asciidoc/router.adoc index fe3692dcef..b139cee2e2 100644 --- a/src/reference/asciidoc/router.adoc +++ b/src/reference/asciidoc/router.adoc @@ -580,9 +580,10 @@ Second, you can define the routing function within the DSL flow itself, as the f @Bean public IntegrationFlow routerFlow2() { return IntegrationFlows.from("routingChannel") - ., String>route(m -> m.getHeaders().get("testHeader", String.class), m -> m - .channelMapping("someHeaderValue", "channelA") - .channelMapping("someOtherHeaderValue", "channelB"), + .route(Message.class, m -> m.getHeaders().get("testHeader", String.class), + m -> m + .channelMapping("someHeaderValue", "channelA") + .channelMapping("someOtherHeaderValue", "channelB"), e -> e.id("headerValueRouter")) .get(); }