GH-8626: Provide cleaner transform() DSL (#8653)

* GH-8626: Provide cleaner `transform()` DSL

Fixes https://github.com/spring-projects/spring-integration/issues/8626

* Add missed `transform(String beanName, @Nullable String methodName)` API
* Introduce a `TransformerSpec` to expose a strict API to configure
transformer variants.
* Introduce `transformWith(Consumer<TransformerSpec>)` as a single point
of all possible transformer and its endpoint options
* Deprecate those `IntegrationFlowDefinition.transform()` variants
which are harder to configure as several lambda arguments

This change will make Kotlin & Groovy DSL more readable and straightforward

* * Use new `transformWith()` in tests where a deprecated API still used
* Add JavaDocs to `TransformerSpec`
* Fix generic types for `BaseIntegrationFlowDefinition.transformWith()` -
they make sense exactly on the `TransformerSpec.transformer()` only
* Apply `transformWith()` for Groovy DSL
* Introduce a new `ClassUtils.isLambda(Object candidate)`
and add a check for Groovy `Closure`
* Fix `GroovyIntegrationFlowDefinition.createConfigurerIfAny()` to propagate a `Consumer` argument
down to the `Closure`

* * Rename `TransformerSpec -> TransformerEndpointSpec` for better context
meaning of the class
* Introduce `KotlinTransformerEndpointSpec` as an extension of the `TransformerEndpointSpec`
to have an `inline fun <reified P> transformer(crossinline function: (P) -> Any)`
for Kotlin style
* Add `KotlinIntegrationFlowDefinition.transformWith(KotlinTransformerEndpointSpec)`
* Deprecate Kotlin methods which are covered by the mentioned `transformWith()`
* Fix tests to use new API
* Mentioned the change in the doc

spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java

@@ -538,12 +538,18 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 	/**
 	 * Populate the {@code Transformer} EI Pattern specific {@link MessageHandler} implementation
 	 * for the SpEL {@link Expression}.
+	 * Shortcut for:
+	 * <pre class="code">
+	 * {@code
+	 *  .transformWith((transformerSpec) -> transformerSpec.expression(expression))
+	 * }
+	 * </pre>
 	 * @param expression the {@code Transformer} {@link Expression}.
 	 * @return the current {@link BaseIntegrationFlowDefinition}.
 	 * @see ExpressionEvaluatingTransformer
 	 */
 	public B transform(String expression) {
-		return transform(expression, (Consumer<GenericEndpointSpec<MessageTransformingHandler>>) null);
+		return transformWith((transformerSpec) -> transformerSpec.expression(expression));
 	}
 
 	/**
@@ -552,8 +558,10 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 	 * @param expression the {@code Transformer} {@link Expression}.
 	 * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options.
 	 * @return the current {@link BaseIntegrationFlowDefinition}.
+	 * @deprecated since 6.2 in favor of {@link #transformWith(Consumer)}.
 	 * @see ExpressionEvaluatingTransformer
 	 */
+	@Deprecated(since = "6.2", forRemoval = true)
 	public B transform(String expression,
 			@Nullable Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
 
@@ -566,9 +574,15 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 	/**
 	 * Populate the {@code MessageTransformingHandler} for the {@link MethodInvokingTransformer}
 	 * to invoke the discovered service method at runtime.
+	 * Shortcut for:
+	 * <pre class="code">
+	 * {@code
+	 *  .transformWith((transformerSpec) -> transformerSpec.ref(service))
+	 * }
+	 * </pre>
 	 * @param service the service to use.
 	 * @return the current {@link BaseIntegrationFlowDefinition}.
-	 * @see ExpressionEvaluatingTransformer
+	 * @see MethodInvokingTransformer
 	 */
 	public B transform(Object service) {
 		return transform(service, null);
@@ -577,13 +591,36 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 	/**
 	 * Populate the {@code MessageTransformingHandler} for the {@link MethodInvokingTransformer}
 	 * to invoke the service method at runtime.
+	 * <pre class="code">
+	 * {@code
+	 *  .transformWith((transformerSpec) -> transformerSpec.ref(service).method(methodName))
+	 * }
+	 * </pre>
 	 * @param service the service to use.
 	 * @param methodName the method to invoke.
 	 * @return the current {@link BaseIntegrationFlowDefinition}.
 	 * @see MethodInvokingTransformer
 	 */
 	public B transform(Object service, @Nullable String methodName) {
-		return transform(service, methodName, null);
+		return transformWith((transformerSpec) -> transformerSpec.ref(service).method(methodName));
+	}
+
+	/**
+	 * Populate the {@code MessageTransformingHandler} for the {@link MethodInvokingTransformer}
+	 * to invoke the bean method at runtime.
+	 * <pre class="code">
+	 * {@code
+	 *  .transformWith((transformerSpec) -> transformerSpec.refName(beanName).method(methodName))
+	 * }
+	 * </pre>
+	 * @param beanName the name for bean to resolve lazily.
+	 * @param methodName the method to invoke.
+	 * @return the current {@link BaseIntegrationFlowDefinition}.
+	 * @since 6.2
+	 * @see MethodInvokingTransformer
+	 */
+	public B transform(String beanName, @Nullable String methodName) {
+		return transformWith((transformerSpec) -> transformerSpec.refName(beanName).method(methodName));
 	}
 
 	/**
@@ -593,8 +630,10 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 	 * @param methodName the method to invoke.
 	 * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options.
 	 * @return the current {@link BaseIntegrationFlowDefinition}.
-	 * @see ExpressionEvaluatingTransformer
+	 * @deprecated since 6.2 in favor of {@link #transformWith(Consumer)}.
+	 * @see MethodInvokingTransformer
 	 */
+	@Deprecated(since = "6.2", forRemoval = true)
 	public B transform(Object service, @Nullable String methodName,
 			@Nullable Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
 
@@ -617,12 +656,18 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 	 *  .transform(Scripts.script("classpath:myScript.py").variable("foo", bar()))
 	 * }
 	 * </pre>
+	 * Shortcut for:
+	 * <pre class="code">
+	 * {@code
+	 *  .transformWith((transformerSpec) -> transformerSpec.processor(messageProcessorSpec))
+	 * }
+	 * </pre>
 	 * @param messageProcessorSpec the {@link MessageProcessorSpec} to use.
 	 * @return the current {@link BaseIntegrationFlowDefinition}.
 	 * @see MethodInvokingTransformer
 	 */
 	public B transform(MessageProcessorSpec<?> messageProcessorSpec) {
-		return transform(messageProcessorSpec, (Consumer<GenericEndpointSpec<MessageTransformingHandler>>) null);
+		return transformWith((transformerSpec) -> transformerSpec.processor(messageProcessorSpec));
 	}
 
 	/**
@@ -638,8 +683,10 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 	 * @param messageProcessorSpec the {@link MessageProcessorSpec} to use.
 	 * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options.
 	 * @return the current {@link BaseIntegrationFlowDefinition}.
+	 * @deprecated since 6.2 in favor of {@link #transformWith(Consumer)}.
 	 * @see MethodInvokingTransformer
 	 */
+	@Deprecated(since = "6.2", forRemoval = true)
 	public B transform(MessageProcessorSpec<?> messageProcessorSpec,
 			@Nullable Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
 
@@ -679,7 +726,8 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 	 * @see LambdaMessageProcessor
 	 */
 	public <P, T> B transform(@Nullable Class<P> expectedType, GenericTransformer<P, T> genericTransformer) {
-		return transform(expectedType, genericTransformer, null);
+		return transformWith((transformerSpec) ->
+				transformerSpec.transformer(genericTransformer).expectedType(expectedType));
 	}
 
 	/**
@@ -714,10 +762,12 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 	 * @param <P> the payload type - 'transform from', or {@code Message.class}.
 	 * @param <T> the target type - 'transform to'.
 	 * @return the current {@link BaseIntegrationFlowDefinition}.
+	 * @deprecated since 6.2 in favor of {@link #transformWith(Consumer)}
 	 * @see MethodInvokingTransformer
 	 * @see LambdaMessageProcessor
 	 * @see GenericEndpointSpec
 	 */
+	@Deprecated(since = "6.2", forRemoval = true)
 	public <P, T> B transform(@Nullable Class<P> expectedType, GenericTransformer<P, T> genericTransformer,
 			@Nullable Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
 
@@ -730,6 +780,16 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 				.handle(new MessageTransformingHandler(transformer), endpointConfigurer);
 	}
 
+	/**
+	 * Populate a {@link MessageTransformingHandler} into the endpoint with provided {@link TransformerEndpointSpec} options.
+	 * One of the 'expression', 'ref', 'refName', 'processor' or 'function' must be provided.
+	 * @return the current {@link BaseIntegrationFlowDefinition}.
+	 * @since 6.2
+	 */
+	public B transformWith(Consumer<TransformerEndpointSpec> transformerConfigurer) {
+		return register(new TransformerEndpointSpec(), transformerConfigurer);
+	}
+
 	/**
 	 * Populate a {@link MessageFilter} with {@link MessageSelector} for the provided SpEL expression.
 	 * @param expression the SpEL expression.
@@ -2787,7 +2847,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 	}
 
 	protected <S extends ConsumerEndpointSpec<? super S, ? extends MessageHandler>> B register(S endpointSpec,
-			@Nullable Consumer<S> endpointConfigurer) {
+			@Nullable Consumer<? super S> endpointConfigurer) {
 
 		if (endpointConfigurer != null) {
 			endpointConfigurer.accept(endpointSpec);

spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java

@@ -50,6 +50,12 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 * Populate the {@link MessageTransformingHandler} instance for the provided
 	 * {@link GenericTransformer}. Use {@link #transform(Class, GenericTransformer)} if
 	 * you need to access the entire message.
+	 * Shortcut for:
+	 * <pre class="code">
+	 * {@code
+	 *  .transformWith((transformerSpec) -> transformerSpec.function(genericTransformer))
+	 * }
+	 * </pre>
 	 * @param genericTransformer the {@link GenericTransformer} to populate.
 	 * @param <S> the source type - 'transform from'.
 	 * @param <T> the target type - 'transform to'.
@@ -58,7 +64,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 * @see org.springframework.integration.handler.LambdaMessageProcessor
 	 */
 	public <S, T> B transform(GenericTransformer<S, T> genericTransformer) {
-		return transform(null, genericTransformer);
+		return transformWith((transformerSpec) -> transformerSpec.transformer(genericTransformer));
 	}
 
 
@@ -66,18 +72,21 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 * Populate the {@link MessageTransformingHandler} instance for the provided
 	 * {@link GenericTransformer}. In addition, accept options for the integration endpoint
 	 * using {@link GenericEndpointSpec}. Use
-	 * {@link #transform(Class, GenericTransformer, Consumer)} if you need to access the
-	 * entire message.
+	 * {@code .transform((transformerSpec) -> transformerSpec.function(genericTransformer).expectedType(Message.class))}
+	 * if you need to access the entire message.
 	 * @param genericTransformer the {@link GenericTransformer} to populate.
 	 * @param endpointConfigurer the {@link Consumer} to provide integration endpoint
 	 * options.
 	 * @param <S> the source type - 'transform from'.
 	 * @param <T> the target type - 'transform to'.
 	 * @return the current {@link IntegrationFlowDefinition}.
+	 * @deprecated since 6.2 in favor of {@link #transformWith(Consumer)}
 	 * @see org.springframework.integration.transformer.MethodInvokingTransformer
 	 * @see org.springframework.integration.handler.LambdaMessageProcessor
 	 * @see GenericEndpointSpec
 	 */
+	@Deprecated(since = "6.2", forRemoval = true)
+	@SuppressWarnings("removal")
 	public <S, T> B transform(GenericTransformer<S, T> genericTransformer,
 			Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
 
@@ -108,7 +117,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 * Populate a {@link org.springframework.integration.filter.MessageFilter}
 	 * with {@link org.springframework.integration.filter.MethodInvokingSelector}
 	 * for the provided {@link GenericSelector}.
-	 * In addition accept options for the integration endpoint using {@link FilterEndpointSpec}.
+	 * In addition, accept options for the integration endpoint using {@link FilterEndpointSpec}.
 	 * Typically used with a Java 8 Lambda expression:
 	 * <pre class="code">
 	 * {@code
@@ -152,7 +161,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 * Populate a {@link ServiceActivatingHandler} for the
 	 * {@link org.springframework.integration.handler.MethodInvokingMessageProcessor}
 	 * to invoke the provided {@link GenericHandler} at runtime.
-	 * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}.
+	 * In addition, accept options for the integration endpoint using {@link GenericEndpointSpec}.
 	 * Typically used with a Java 8 Lambda expression:
 	 * <pre class="code">
 	 * {@code
@@ -177,7 +186,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	/**
 	 * Populate the {@link MethodInvokingSplitter} to evaluate the provided
 	 * {@link Function} at runtime.
-	 * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}.
+	 * In addition, accept options for the integration endpoint using {@link GenericEndpointSpec}.
 	 * Typically used with a Java 8 Lambda expression:
 	 * <pre class="code">
 	 * {@code

spring-integration-core/src/main/java/org/springframework/integration/dsl/TransformerEndpointSpec.java

@@ -0,0 +1,227 @@
+/*
+ * Copyright 2023-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.dsl;
+
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.springframework.expression.Expression;
+import org.springframework.integration.core.GenericTransformer;
+import org.springframework.integration.handler.BeanNameMessageProcessor;
+import org.springframework.integration.handler.LambdaMessageProcessor;
+import org.springframework.integration.handler.MessageProcessor;
+import org.springframework.integration.transformer.ExpressionEvaluatingTransformer;
+import org.springframework.integration.transformer.MessageTransformingHandler;
+import org.springframework.integration.transformer.MethodInvokingTransformer;
+import org.springframework.integration.transformer.Transformer;
+import org.springframework.integration.util.ClassUtils;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * A {@link ConsumerEndpointSpec} for a {@link MessageTransformingHandler} options.
+ * One of the {@link #expression(String)}, {@link #ref(Object)}, {@link #refName(String)},
+ * {@link #processor(MessageProcessorSpec)} or {@link #transformer(GenericTransformer)} must be provided.
+ *
+ * @author Artem Bilan
+ *
+ * @since 6.2
+ */
+public class TransformerEndpointSpec extends ConsumerEndpointSpec<TransformerEndpointSpec, MessageTransformingHandler> {
+
+	private final AtomicBoolean transformerSet = new AtomicBoolean();
+
+	private Expression expression;
+
+	private Object ref;
+
+	private String refName;
+
+	@Nullable
+	private String method;
+
+	private GenericTransformer<?, ?> transformer;
+
+	@Nullable
+	private Class<?> expectedType;
+
+	private MessageProcessorSpec<?> processor;
+
+	protected TransformerEndpointSpec() {
+		super(new MessageTransformingHandler());
+	}
+
+	/**
+	 * Provide an expression to use an {@link ExpressionEvaluatingTransformer} for the target handler.
+	 * @param expression the SpEL expression to use.
+	 * @return the TransformerSpec
+	 */
+	public TransformerEndpointSpec expression(String expression) {
+		return expression(PARSER.parseExpression(expression));
+	}
+
+	/**
+	 * Provide an expression to use an {@link ExpressionEvaluatingTransformer} for the target handler.
+	 * @param expression the SpEL expression to use.
+	 * @return the TransformerSpec
+	 */
+	public TransformerEndpointSpec expression(Expression expression) {
+		assertTransformerSet();
+		this.expression = expression;
+		return this;
+	}
+
+	/**
+	 * Provide a service to use a {@link MethodInvokingTransformer} for the target handler.
+	 * @param ref the service to call as a transformer POJO.
+	 * @return the TransformerSpec
+	 */
+	public TransformerEndpointSpec ref(Object ref) {
+		assertTransformerSet();
+		this.ref = ref;
+		return this;
+	}
+
+	/**
+	 * Provide a bean name to use a {@link MethodInvokingTransformer}
+	 * (based on {@link BeanNameMessageProcessor}) for the target handler.
+	 * @param refName the bean name for service to call as a transformer POJO.
+	 * @return the TransformerSpec
+	 */
+	public TransformerEndpointSpec refName(String refName) {
+		assertTransformerSet();
+		this.refName = refName;
+		return this;
+	}
+
+	/**
+	 * Provide a service method name to call. Optional.
+	 * Use only together with {@link #ref(Object)} or {@link #refName(String)}.
+	 * @param method the service method name to call.
+	 * @return the TransformerSpec
+	 */
+	public TransformerEndpointSpec method(@Nullable String method) {
+		this.method = method;
+		return this;
+	}
+
+	/**
+	 * Provide a {@link GenericTransformer} as a direct delegate for {@link MessageTransformingHandler}.
+	 * @param transformer the {@link GenericTransformer} instance to use.
+	 * @param <P> the input type.
+	 * @param <T> the output type.
+	 * @return the TransformerSpec
+	 */
+	public <P, T> TransformerEndpointSpec transformer(GenericTransformer<P, T> transformer) {
+		assertTransformerSet();
+		this.transformer = transformer;
+		return this;
+	}
+
+	/**
+	 * Set a {@link GenericTransformer} input argument type.
+	 * Can be a {@link org.springframework.messaging.Message}.
+	 * Ignored for all other transformers, but {@link #transformer(GenericTransformer)}.
+	 * @param expectedType the {@link GenericTransformer} input argument type.
+	 * @param <P> the type ot expect.
+	 * @return the spec.
+	 */
+	public <P> TransformerEndpointSpec expectedType(@Nullable Class<P> expectedType) {
+		this.expectedType = expectedType;
+		return this;
+	}
+
+	/**
+	 * Provide a {@link MessageProcessorSpec} as a factory for {@link MethodInvokingTransformer} delegate.
+	 * @param processor the {@link MessageProcessorSpec} to use.
+	 * @return the TransformerSpec
+	 */
+	public TransformerEndpointSpec processor(MessageProcessorSpec<?> processor) {
+		assertTransformerSet();
+		this.processor = processor;
+		return this;
+	}
+
+	private void assertTransformerSet() {
+		Assert.isTrue(this.transformerSet.compareAndSet(false, true), this::assertMessage);
+	}
+
+	private String assertMessage() {
+		String currentTransformerValue = null;
+		if (this.expression != null) {
+			currentTransformerValue = "'expression'=" + this.expression;
+		}
+		else if (this.ref != null) {
+			currentTransformerValue = "'ref'=" + this.ref;
+		}
+		else if (this.refName != null) {
+			currentTransformerValue = "'refName'=" + this.refName;
+		}
+		else if (this.transformer != null) {
+			currentTransformerValue = "'transformer'=" + this.transformer;
+		}
+		else if (this.processor != null) {
+			currentTransformerValue = "'processor'=" + this.processor;
+		}
+		return "Only one of the 'expression', 'ref', 'refName', 'processor' or 'transformer' can be set. " +
+				"Current one is " + currentTransformerValue;
+	}
+
+	@Override
+	public Map<Object, String> getComponentsToRegister() {
+		Transformer transformer;
+		if (this.expression != null) {
+			transformer = new ExpressionEvaluatingTransformer(this.expression);
+		}
+		else if (this.ref != null) {
+			if (this.method != null) {
+				transformer = new MethodInvokingTransformer(this.ref, this.method);
+			}
+			else {
+				transformer = new MethodInvokingTransformer(this.ref);
+			}
+		}
+		else if (this.refName != null) {
+			transformer = new MethodInvokingTransformer(new BeanNameMessageProcessor<>(this.refName, this.method));
+		}
+		else if (this.processor != null) {
+			MessageProcessor<?> targetProcessor = this.processor.getObject();
+			this.componentsToRegister.put(targetProcessor, null);
+			transformer = new MethodInvokingTransformer(targetProcessor);
+		}
+		else if (this.transformer != null) {
+			transformer = wrapToTransformerIfAny();
+		}
+		else {
+			throw new IllegalStateException(
+					"One of the 'expression', 'ref', 'refName', 'processor' or 'transformer' must be provided.");
+		}
+
+		this.handler.setTransformer(transformer);
+
+		this.componentsToRegister.put(transformer, null);
+		return super.getComponentsToRegister();
+	}
+
+	private Transformer wrapToTransformerIfAny() {
+		return this.transformer instanceof Transformer ? (Transformer) this.transformer :
+				(ClassUtils.isLambda(this.transformer)
+						? new MethodInvokingTransformer(new LambdaMessageProcessor(this.transformer, this.expectedType))
+						: new MethodInvokingTransformer(this.transformer, ClassUtils.TRANSFORMER_TRANSFORM_METHOD));
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -41,7 +41,16 @@ import org.springframework.util.Assert;
  */
 public class MessageTransformingHandler extends AbstractReplyProducingMessageHandler implements ManageableLifecycle {
 
-	private final Transformer transformer;
+	private Transformer transformer;
+
+	/**
+	 * Create a {@link MessageTransformingHandler} instance.
+	 * The target delegate {@link Transformer} must be provided then via setter.
+	 * @since 6.2
+	 */
+	public MessageTransformingHandler() {
+		setRequiresReply(true);
+	}
 
 	/**
 	 * Create a {@link MessageTransformingHandler} instance that delegates to
@@ -49,11 +58,20 @@ public class MessageTransformingHandler extends AbstractReplyProducingMessageHan
 	 * @param transformer The transformer.
 	 */
 	public MessageTransformingHandler(Transformer transformer) {
+		this();
 		Assert.notNull(transformer, "transformer must not be null");
 		this.transformer = transformer;
-		this.setRequiresReply(true);
 	}
 
+	/**
+	 * Provide a {@link Transformer} delegate.
+	 * @param transformer the {@link Transformer} to use.
+	 * @since 6.2
+	 */
+	public void setTransformer(Transformer transformer) {
+		Assert.notNull(transformer, "transformer must not be null");
+		this.transformer = transformer;
+	}
 
 	@Override
 	public String getComponentType() {
@@ -77,6 +95,7 @@ public class MessageTransformingHandler extends AbstractReplyProducingMessageHan
 
 	@Override
 	protected void doInit() {
+		Assert.notNull(this.transformer, "transformer must not be null");
 		BeanFactory beanFactory = getBeanFactory();
 		if (beanFactory != null && this.transformer instanceof BeanFactoryAware) {
 			((BeanFactoryAware) this.transformer).setBeanFactory(beanFactory);

spring-integration-core/src/main/java/org/springframework/integration/util/ClassUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,6 +17,7 @@
 package org.springframework.integration.util;
 
 import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
@@ -223,6 +224,19 @@ public abstract class ClassUtils {
 		return PRIMITIVE_WRAPPER_TYPE_MAP.get(clazz);
 	}
 
+	/**
+	 * Check if object is Java, Kotlin or Groovy lambda.
+	 * @param candidate the {@link Object} to check.
+	 * @return true if object is a Java, Kotlin or Groovy lambda.
+	 * @since 6.2
+	 */
+	public static boolean isLambda(Object candidate) {
+		Class<?> aClass = candidate.getClass();
+		return isLambda(aClass) ||
+				(Proxy.isProxyClass(aClass)  // Groovy Closure is a Lambda in Java terms
+						&& Proxy.getInvocationHandler(candidate).getClass().getSimpleName().equals("ConvertedClosure"));
+	}
+
 	/**
 	 * Check if class is Java or Kotlin lambda.
 	 * @param aClass the {@link Class} to check.

spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt

@@ -92,9 +92,16 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
 	}
 
 	/**
-	 * Inline function for [IntegrationFlowDefinition.transform] providing a `transform<MyTypeIn, MyTypeOut>()` variant
+	 * Inline function for [IntegrationFlowDefinition.transform] providing a `transform<MyTypeIn>()` variant
 	 * with reified generic type.
 	 */
+	@Deprecated("since 6.2",
+			ReplaceWith("""
+				transformWith { 
+					transformer<Type> { }
+					id("value")
+				}"""))
+	@Suppress("DEPRECATION", "REMOVAL")
 	inline fun <reified P> transform(
 		crossinline function: (P) -> Any,
 		crossinline configurer: GenericEndpointSpec<MessageTransformingHandler>.() -> Unit
@@ -103,6 +110,16 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
 		this.delegate.transform(P::class.java, { function(it) }) { configurer(it) }
 	}
 
+	/**
+	 * Inline function for [IntegrationFlowDefinition.transformWith]
+	 * providing a `transform<MyTypeIn>()` variant
+	 * with reified generic type.
+	 * @since 6.2
+	 */
+	fun transformWith(configurer: KotlinTransformerEndpointSpec.() -> Unit) {
+		this.delegate.register(KotlinTransformerEndpointSpec(), configurer)
+	}
+
 	/**
 	 * Inline function for [IntegrationFlowDefinition.split] providing a `split<MyTypeIn>()` variant
 	 * with reified generic type.
@@ -305,6 +322,13 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
 	 * for the provided `Transformer` instance.
 	 * @since 5.3.1
 	 */
+	@Deprecated("since 6.2",
+			ReplaceWith("""
+				transformWith { 
+					transformer(transformer)
+					id("value")
+				}"""))
+	@Suppress("DEPRECATION", "REMOVAL")
 	fun transform(
 		transformer: Transformer,
 		endpointConfigurer: GenericEndpointSpec<MessageTransformingHandler>.() -> Unit = {}
@@ -313,10 +337,26 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
 		this.delegate.transform(transformer) { endpointConfigurer(it) }
 	}
 
+	/**
+	 * Populate the [Transformer] EI Pattern specific [MessageHandler] implementation
+	 * for the provided [Transformer] instance.
+	 * @since 6.2
+	 */
+	fun transform(transformer: Transformer) {
+		this.delegate.transform(transformer)
+	}
+
 	/**
 	 * Populate the [Transformer] EI Pattern specific [MessageHandler] implementation
 	 * for the SpEL [Expression].
 	 */
+	@Deprecated("since 6.2",
+			ReplaceWith("""
+				transformWith { 
+					expression("value")
+					id("value")
+				}"""))
+	@Suppress("DEPRECATION", "REMOVAL")
 	fun transform(
 		expression: String,
 		endpointConfigurer: GenericEndpointSpec<MessageTransformingHandler>.() -> Unit = {}
@@ -325,6 +365,16 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
 		this.delegate.transform(expression, endpointConfigurer)
 	}
 
+
+	/**
+	 * Populate the [Transformer] EI Pattern specific [MessageHandler] implementation
+	 * for the SpEL [Expression].
+	 * @since 6.2
+	 */
+	fun transform(expression: String) {
+		this.delegate.transform(expression)
+	}
+
 	/**
 	 * Populate the [MessageTransformingHandler] for the [MethodInvokingTransformer]
 	 * to invoke the service method at runtime.
@@ -337,6 +387,14 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
 	 * Populate the [MessageTransformingHandler] for the [MethodInvokingTransformer]
 	 * to invoke the service method at runtime.
 	 */
+	@Deprecated("since 6.2",
+			ReplaceWith("""
+				transformWith { 
+					ref("value")
+					method("value")
+					id("value")
+				}"""))
+	@Suppress("DEPRECATION", "REMOVAL")
 	fun transform(
 		service: Any, methodName: String?,
 		endpointConfigurer: GenericEndpointSpec<MessageTransformingHandler>.() -> Unit
@@ -350,6 +408,13 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
 	 * [org.springframework.integration.handler.MessageProcessor] from provided [MessageProcessorSpec].
 	 * In addition, accept options for the integration endpoint using [GenericEndpointSpec].
 	 */
+	@Deprecated("since 6.2",
+			ReplaceWith("""
+				transformWith { 
+					processor("value")
+					id("value")
+				}"""))
+	@Suppress("DEPRECATION", "REMOVAL")
 	fun transform(
 		messageProcessorSpec: MessageProcessorSpec<*>,
 		endpointConfigurer: GenericEndpointSpec<MessageTransformingHandler>.() -> Unit = {}
@@ -358,6 +423,15 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
 		this.delegate.transform(messageProcessorSpec, endpointConfigurer)
 	}
 
+	/**
+	 * Populate the [MessageTransformingHandler] instance for the
+	 * [org.springframework.integration.handler.MessageProcessor] from provided [MessageProcessorSpec].
+	 * @since 6.2
+	 */
+	fun transform(messageProcessorSpec: MessageProcessorSpec<*>) {
+		this.delegate.transform(messageProcessorSpec)
+	}
+
 	/**
 	 * Populate a [MessageFilter] with [MessageSelector] for the provided SpEL expression.
 	 * In addition, accept options for the integration endpoint using [KotlinFilterEndpointSpec]:
@@ -566,7 +640,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
 				delay { 
 					messageGroupId(groupId) 
 				}"""))
-	@Suppress("DEPRECATION")
+	@Suppress("DEPRECATION", "REMOVAL")
 	fun delay(groupId: String, endpointConfigurer: DelayerEndpointSpec.() -> Unit = {}) {
 		this.delegate.delay(groupId, endpointConfigurer)
 	}

spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinTransformerEndpointSpec.kt

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.dsl
+
+import org.springframework.integration.transformer.MessageTransformingHandler
+
+/**
+ * A [TransformerEndpointSpec] wrapped for Kotlin DSL.
+ *
+ * @property delegate the [TransformerEndpointSpec] this instance is delegating to.
+ *
+ * @author Artem Bilan
+ *
+ * @since 6.2
+ */
+class KotlinTransformerEndpointSpec : TransformerEndpointSpec() {
+
+	/**
+	 * Provide a Kotlin function as a direct delegate for [MessageTransformingHandler].
+	 * @param function the function instance to use.
+	 * @param <P> the input type.
+	 */
+	inline fun <reified P> transformer(crossinline function: (P) -> Any) {
+		expectedType(P::class.java)
+		transformer<P, Any> { function(it) }
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java

@@ -64,6 +64,7 @@ import org.springframework.integration.dsl.MessageChannels;
 import org.springframework.integration.dsl.PollerSpec;
 import org.springframework.integration.dsl.Pollers;
 import org.springframework.integration.dsl.QueueChannelSpec;
+import org.springframework.integration.dsl.TransformerEndpointSpec;
 import org.springframework.integration.dsl.Transformers;
 import org.springframework.integration.endpoint.AbstractEndpoint;
 import org.springframework.integration.endpoint.EventDrivenConsumer;
@@ -671,17 +672,24 @@ public class IntegrationFlowTests {
 					.fixedSubscriberChannel()
 					.<String, Integer>transform(Integer::parseInt)
 					.transform(Foo::new)
-					.transform(new PayloadSerializingTransformer(),
-							c -> c.autoStartup(false).id("payloadSerializingTransformer"))
+					.transformWith(this::payloadSerializingTransformer)
 					.channel(MessageChannels.queue(new SimpleMessageStore(), "fooQueue"))
 					.transform(Transformers.deserializer(Foo.class.getName()))
 					.<Foo, Integer>transform(f -> f.value)
 					.filter("true", e -> e.id("expressionFilter"))
 					.channel(publishSubscribeChannel())
-					.transform((Integer p) -> p * 2, c -> c.advice(this.expressionAdvice()))
+					.transformWith(t -> t
+							.transformer((Integer p) -> p * 2)
+							.advice(expressionAdvice()))
 					.get();
 		}
 
+		private void payloadSerializingTransformer(TransformerEndpointSpec spec) {
+			spec.transformer(new PayloadSerializingTransformer())
+					.autoStartup(false)
+					.id("payloadSerializingTransformer");
+		}
+
 		@Bean
 		public MessageChannel publishSubscribeChannel() {
 			return new PublishSubscribeChannel();

spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2022 the original author or authors.
+ * Copyright 2016-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -190,7 +190,8 @@ public class ManualFlowTests {
 		IntegrationFlow myFlow = f -> f
 				.<String, String>transform(String::toUpperCase)
 				.channel(MessageChannels.queue())
-				.transform("Hello, "::concat, e -> e
+				.transformWith(t -> t
+						.transformer("Hello, "::concat)
 						.poller(p -> p
 								.fixedDelay(10)
 								.maxMessagesPerPoll(1)

spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2022 the original author or authors.
+ * Copyright 2016-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -262,8 +262,10 @@ public class ReactiveStreamsTests {
 			return IntegrationFlow
 					.from("inputChannel")
 					.split(s -> s.delimiters(","))
-					.<String, Integer>transform(Integer::parseInt,
-							e -> e.reactive(flux -> flux.publishOn(Schedulers.parallel())).id("reactiveTransformer"))
+					.transformWith(t -> t
+							.<String, Integer>transformer(Integer::parseInt)
+							.reactive(flux -> flux.publishOn(Schedulers.parallel()))
+							.id("reactiveTransformer"))
 					.channel(MessageChannels.queue())
 					.log()
 					.toReactivePublisher();

spring-integration-core/src/test/java/org/springframework/integration/dsl/transformers/TransformerTests.java

@@ -460,10 +460,11 @@ public class TransformerTests {
 		@Bean
 		public IntegrationFlow transformFlowWithError() {
 			return f -> f
-					.transform(p -> {
-								throw new RuntimeException("intentional");
-							},
-							e -> e.advice(expressionAdvice()))
+					.transformWith((t) ->
+							t.transformer(p -> {
+										throw new RuntimeException("intentional");
+									})
+									.advice(expressionAdvice()))
 					.log();
 		}
 

spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/KotlinDslTests.kt

@@ -161,7 +161,10 @@ class KotlinDslTests {
 
 		val integrationFlow =
 			integrationFlow(publisher) {
-				transform<Message<Int>>({ it.payload * 2 }) { id("foo") }
+				transformWith {
+					transformer<Message<Int>> { it.payload * 2 }
+					id("foo")
+				}
 				channel(fluxChannel)
 			}
 
@@ -249,7 +252,10 @@ class KotlinDslTests {
 		@Bean
 		fun functionFlow() =
 			integrationFlow<Function<ByteArray, String>>({ beanName("functionGateway") }) {
-				transform(Transformers.objectToString()) { id("objectToStringTransformer") }
+				transformWith {
+					transformer(Transformers.objectToString())
+					id("objectToStringTransformer")
+				}
 				transform<String> { it.uppercase() }
 				split<Message<*>> { it.payload }
 				split<String>({ it }) { id("splitterEndpoint") }
@@ -292,7 +298,10 @@ class KotlinDslTests {
 		fun fixedSubscriberFlow() =
 			integrationFlow("fixedSubscriberInput", true) {
 				log<Any>(LoggingHandler.Level.WARN) { it.payload }
-				transform("payload") { id("spelTransformer") }
+				transformWith {
+					expression("payload")
+					id("spelTransformer")
+				}
 			}
 
 		@Bean