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 e3b05a7a14..de364b9c29 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 @@ -460,7 +460,7 @@ public abstract class IntegrationFlowDefinition> endpointConfigurer) { - return this.handle(new ServiceActivatingHandler(new ExpressionCommandMessageProcessor( + return handle(new ServiceActivatingHandler(new ExpressionCommandMessageProcessor( new ControlBusMethodFilter())), endpointConfigurer); } @@ -537,7 +537,9 @@ public abstract class IntegrationFlowDefinition the source type - 'transform from'. * @param the target type - 'transform to'. @@ -599,16 +601,22 @@ public abstract class IntegrationFlowDefinition B convert(Class

payloadType) { + Assert.isTrue(!payloadType.equals(Message.class), ".convert() does not support Message as an explicit type"); return transform(payloadType, p -> p); } /** - * Populate the {@link MessageTransformingHandler} instance for the provided {@link GenericTransformer} - * for the specific {@code payloadType} to convert at runtime. - * @param payloadType the {@link Class} for expected payload type. + * Populate the {@link MessageTransformingHandler} instance for the provided + * {@link GenericTransformer} for the specific {@code payloadType} to convert at + * runtime. + * Use {@link #transform(Class, GenericTransformer)} if you need access to the + * entire message. + * @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 transformer. + * Conversion to this type will be attempted, if necessary. * @param genericTransformer the {@link GenericTransformer} to populate. - * @param

the payload type - 'transform from'. + * @param

the payload type - 'transform from' or {@code Message.class}. * @param the target type - 'transform to'. * @return the current {@link IntegrationFlowDefinition}. * @see MethodInvokingTransformer @@ -619,10 +627,14 @@ public abstract class IntegrationFlowDefinition the source type - 'transform from'. * @param the target type - 'transform to'. * @return the current {@link IntegrationFlowDefinition}. @@ -632,7 +644,8 @@ public abstract class IntegrationFlowDefinition B transform(GenericTransformer genericTransformer, Consumer> endpointConfigurer) { - return this.transform(null, genericTransformer, endpointConfigurer); + + return transform(null, genericTransformer, endpointConfigurer); } /** @@ -651,6 +664,7 @@ public abstract class IntegrationFlowDefinition B convert(Class

payloadType, Consumer> endpointConfigurer) { + Assert.isTrue(!payloadType.equals(Message.class), ".convert() does not support Message"); return transform(payloadType, p -> p, endpointConfigurer); } @@ -658,10 +672,12 @@ public abstract class IntegrationFlowDefinition the payload type - 'transform from'. + * @param

the payload type - 'transform from', or {@code Message.class}. * @param the target type - 'transform to'. * @return the current {@link IntegrationFlowDefinition}. * @see MethodInvokingTransformer @@ -758,6 +774,8 @@ public abstract class IntegrationFlowDefinition + * Use {@link #filter(Class, GenericSelector)} if you need to access the entire + * message. * @param genericSelector the {@link GenericSelector} to use. * @param

the source payload type. * @return the current {@link IntegrationFlowDefinition}. @@ -813,14 +831,16 @@ public abstract class IntegrationFlowDefinition p.after(new Date())) * } * - * @param payloadType the {@link Class} for desired {@code payload} type. + * @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 selector. + * Conversion to this type will be attempted, if necessary. * @param genericSelector the {@link GenericSelector} to use. - * @param

the source payload type. + * @param

the source payload type or {@code Message.class}. * @return the current {@link IntegrationFlowDefinition}. * @see LambdaMessageProcessor */ public

B filter(Class

payloadType, GenericSelector

genericSelector) { - return this.filter(payloadType, genericSelector, null); + return filter(payloadType, genericSelector, null); } /** @@ -833,6 +853,8 @@ public abstract class IntegrationFlowDefinition e.autoStartup(false)) * } * + * Use {@link #filter(Class, GenericSelector, Consumer)} if you need to access the entire + * message. * @param genericSelector the {@link GenericSelector} to use. * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. * @param

the source payload type. @@ -853,10 +875,12 @@ public abstract class IntegrationFlowDefinition p.after(new Date()), e -> e.autoStartup(false)) * } * - * @param payloadType the {@link Class} for desired {@code payload} type. + * @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 selector. + * Conversion to this type will be attempted, if necessary. * @param genericSelector the {@link GenericSelector} to use. * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. - * @param

the source payload type. + * @param

the source payload type or {@code Message.class}. * @return the current {@link IntegrationFlowDefinition}. * @see LambdaMessageProcessor * @see FilterEndpointSpec @@ -914,7 +938,7 @@ public abstract class IntegrationFlowDefinitionhandle((p, h) -> p / 2) * } * + * Use {@link #handle(Class, GenericHandler)} if you need to access the entire + * message. * @param handler the handler to invoke. * @param

the payload type to expect. * @return the current {@link IntegrationFlowDefinition}. @@ -1009,6 +1035,8 @@ public abstract class IntegrationFlowDefinitionhandle((p, h) -> p / 2, e -> e.autoStartup(false)) * } * + * Use {@link #handle(Class, GenericHandler, Consumer)} if you need to access the entire + * message. * @param handler the handler to invoke. * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. * @param

the payload type to expect. @@ -1018,7 +1046,7 @@ public abstract class IntegrationFlowDefinition B handle(GenericHandler

handler, Consumer> endpointConfigurer) { - return this.handle(null, handler, endpointConfigurer); + return handle(null, handler, endpointConfigurer); } /** @@ -1031,15 +1059,16 @@ public abstract class IntegrationFlowDefinition p / 2) * } * - * @param payloadType the expected payload type. - * The accepted payload can be converted to this one at runtime + * @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 handler. + * Conversion to this type will be attempted, if necessary. * @param handler the handler to invoke. - * @param

the payload type to expect. + * @param

the payload type to expect, or {@code Message.class}. * @return the current {@link IntegrationFlowDefinition}. * @see LambdaMessageProcessor */ public

B handle(Class

payloadType, GenericHandler

handler) { - return this.handle(payloadType, handler, null); + return handle(payloadType, handler, null); } /** @@ -1053,11 +1082,12 @@ public abstract class IntegrationFlowDefinition p / 2, e -> e.autoStartup(false)) * } * - * @param payloadType the expected payload type. - * The accepted payload can be converted to this one at runtime + * @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 handler. + * Conversion to this type will be attempted, if necessary. * @param handler the handler to invoke. * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. - * @param

the payload type to expect. + * @param

the payload type to expect or {@code Message.class}. * @return the current {@link IntegrationFlowDefinition}. * @see LambdaMessageProcessor */ @@ -1070,7 +1100,7 @@ public abstract class IntegrationFlowDefinition>) null); + return split((Consumer>) null); } /** @@ -1348,7 +1378,7 @@ public abstract class IntegrationFlowDefinition> endpointConfigurer) { - return this.split(new DefaultMessageSplitter(), endpointConfigurer); + return split(new DefaultMessageSplitter(), endpointConfigurer); } /** @@ -1432,7 +1462,7 @@ public abstract class IntegrationFlowDefinition - * @param payloadType the expected payload type. Used at runtime to convert received payload type to. + * @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. + * Conversion to this type will be attempted, if necessary. * @param splitter the splitter {@link Function}. - * @param

the payload type. + * @param

the payload type or {@code Message.class}. * @return the current {@link IntegrationFlowDefinition}. * @see LambdaMessageProcessor */ @@ -1562,20 +1594,23 @@ public abstract class IntegrationFlowDefinition e.applySequence(false)) * } * - * @param payloadType the expected payload type. Used at runtime to convert received payload type to. + * @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. + * Conversion to this type will be attempted, if necessary. * @param splitter the splitter {@link Function}. * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. - * @param

the payload type. + * @param

the payload type or {@code Message.class}. * @return the current {@link IntegrationFlowDefinition}. * @see LambdaMessageProcessor * @see SplitterEndpointSpec */ public

B split(Class

payloadType, Function splitter, Consumer> endpointConfigurer) { + MethodInvokingSplitter split = isLambda(splitter) ? new MethodInvokingSplitter(new LambdaMessageProcessor(splitter, payloadType)) : new MethodInvokingSplitter(splitter, ClassUtils.FUNCTION_APPLY_METHOD); - return this.split(split, endpointConfigurer); + return split(split, endpointConfigurer); } /** @@ -1665,7 +1700,7 @@ public abstract class IntegrationFlowDefinition> endpointConfigurer) { - return this.transform(headerFilter, endpointConfigurer); + return transform(headerFilter, endpointConfigurer); } /** @@ -1689,7 +1724,7 @@ public abstract class IntegrationFlowDefinition> endpointConfigurer) { - return this.transform(new ClaimCheckInTransformer(messageStore), endpointConfigurer); + return transform(new ClaimCheckInTransformer(messageStore), endpointConfigurer); } /** @@ -1730,7 +1765,7 @@ public abstract class IntegrationFlowDefinition> endpointConfigurer) { ClaimCheckOutTransformer claimCheckOutTransformer = new ClaimCheckOutTransformer(messageStore); claimCheckOutTransformer.setRemoveMessage(removeMessage); - return this.transform(claimCheckOutTransformer, endpointConfigurer); + return transform(claimCheckOutTransformer, endpointConfigurer); } /** @@ -1895,6 +1930,7 @@ public abstract class IntegrationFlowDefinition p.equals("foo") || p.equals("bar") ? new String[] {"foo", "bar"} : null) * } * + * Use {@link #route(Class, Function)} if you need to access the entire message. * @param router the {@link Function} to use. * @param the source payload type. * @param the target result type. @@ -1913,9 +1949,11 @@ public abstract class IntegrationFlowDefinition p % 2 == 0) * } * - * @param payloadType the expected payload type. + * @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. + * Conversion to this type will be attempted, if necessary. * @param router the {@link Function} to use. - * @param the source payload type. + * @param the source payload type or {@code Message.class}. * @param the target result type. * @return the current {@link IntegrationFlowDefinition}. * @see LambdaMessageProcessor @@ -1938,6 +1976,7 @@ public abstract class IntegrationFlowDefinition + * Use {@link #route(Class, Function, Consumer)} if you need to access the entire message. * @param router the {@link Function} to use. * @param routerConfigurer the {@link Consumer} to provide {@link MethodInvokingRouter} options. * @param the source payload type. @@ -1962,16 +2001,19 @@ public abstract class IntegrationFlowDefinition - * @param payloadType the expected payload type. + * @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. + * 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. - * @param

the source payload type. + * @param

the source payload type or {@code Message.class}. * @param the target result type. * @return the current {@link IntegrationFlowDefinition}. * @see LambdaMessageProcessor */ public B route(Class

payloadType, Function router, Consumer> routerConfigurer) { + MethodInvokingRouter methodInvokingRouter = isLambda(router) ? new MethodInvokingRouter(new LambdaMessageProcessor(router, payloadType)) : new MethodInvokingRouter(router, ClassUtils.FUNCTION_APPLY_METHOD); 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 12d1dbb1c9..aee6e80439 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 @@ -22,6 +22,9 @@ import java.lang.reflect.Modifier; import java.util.Map; import java.util.concurrent.atomic.AtomicReference; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; @@ -38,11 +41,14 @@ import org.springframework.util.ReflectionUtils; * - functional interface implementations. * * @author Artem Bilan + * @author Gary Russell * * @since 5.0 */ public class LambdaMessageProcessor implements MessageProcessor, BeanFactoryAware { + private static final Log logger = LogFactory.getLog(LambdaMessageProcessor.class); + private final Object target; private final Method method; @@ -124,6 +130,12 @@ public class LambdaMessageProcessor implements MessageProcessor, BeanFac return this.method.invoke(this.target, args); } catch (InvocationTargetException e) { + if (e.getTargetException() instanceof ClassCastException) { + logger.error("Could not invoke the method due to a class cast exception, if using a lambda in the DSL, " + + "consider using an overloaded EIP method that takes a Class argument to explicitly " + + "specify the type. An example of when this often occurs is if the lambda is configured to " + + "receive a Message argument.", e.getCause()); + } throw new MessageHandlingException(message, e.getCause()); } catch (Exception e) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/LambdaMessageProcessorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/LambdaMessageProcessorTests.java index 4f4de4c66c..95769d749a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/LambdaMessageProcessorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/LambdaMessageProcessorTests.java @@ -16,6 +16,7 @@ package org.springframework.integration.dsl; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.hamcrest.Matchers.instanceOf; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThat; @@ -72,6 +73,15 @@ public class LambdaMessageProcessorTests { assertSame(testMessage, result); } + @Test + public void testMessageAsArgumentLambda() { + LambdaMessageProcessor lmp = new LambdaMessageProcessor( + (GenericTransformer, Message>) source -> messageTransformer(source), null); + lmp.setBeanFactory(mock(BeanFactory.class)); + GenericMessage testMessage = new GenericMessage<>("foo"); + assertThatThrownBy(() -> lmp.processMessage(testMessage)).hasCauseExactlyInstanceOf(ClassCastException.class); + } + private void handle(GenericHandler h) { LambdaMessageProcessor lmp = new LambdaMessageProcessor(h, String.class); lmp.setBeanFactory(getBeanFactory()); diff --git a/src/reference/asciidoc/dsl.adoc b/src/reference/asciidoc/dsl.adoc index 48300c97fe..6527c37ab1 100644 --- a/src/reference/asciidoc/dsl.adoc +++ b/src/reference/asciidoc/dsl.adoc @@ -113,6 +113,30 @@ The flow is "'one way'". That is, it does not provide a reply message but only prints the payload to STDOUT. The endpoints are automatically wired together by using direct channels. +[[java-dsl-class-cast]] +.Lambdas And `Message` Arguments +IMPORTANT: When using lambdas in EIP methods, the "input" argument is generally the message payload. +If you wish to access the entire message, use one of the overloaded methods that take a `Class` as the first parameter. +For example, this won't work: + +==== +[source, java] +---- +., Foo>transform(m -> newFooFromMessage(m)) +---- +==== + +This will fail at runtime with a `ClassCastException` because the lambda doesn't retain the argument type and the framework will attempt to cast the payload to a `Message`. + +Instead, use: + +==== +[source, java] +---- +.(Message.class, m -> newFooFromMessage(m)) +---- +==== + [[java-dsl-channels]] === Message Channels @@ -294,6 +318,8 @@ Nevertheless, the DSL parser takes care of bean declarations for inline objects, See [https://docs.spring.io/spring-integration/api/org/springframework/integration/dsl/Transformers.html] in the Javadoc for more information and supported factory methods. +Also see <>. + [[java-dsl-inbound-adapters]] === Inbound Channel Adapters @@ -403,6 +429,8 @@ public IntegrationFlow recipientListFlow() { The `.defaultOutputToParentFlow()` of the `.routeToRecipients()` definition lets you set the router's `defaultOutput` as a gateway to continue a process for the unmatched messages in the main flow. +Also see <>. + [[java-dsl-splitters]] === Splitters @@ -427,6 +455,8 @@ public IntegrationFlow splitFlow() { The preceding example creates a splitter that splits a message containing a comma-delimited `String`. Note: The `getT2()` method comes from a `Tuple` `Collection`, which is the result of `EndpointSpec.get()`, and represents a pair of `ConsumerEndpointFactoryBean` and `DefaultMessageSplitter` for the preceding example. +Also see <>. + [[java-dsl-aggregators]] === Aggregators and Resequencers @@ -531,6 +561,8 @@ public IntegrationFlow integerFlow() { ---- ==== +Also see <>. + [[java-dsl-log]] === Operator log()