Optimize MessageTriggerAction for Java DSL (spring-projects#8595)

Currently, the `MessageTriggerAction.trigger` is configured
on the `BaseIntegrationFlowDefinition` to be called via reflection
in the `MessagingMethodInvokerHelper`

* Represent a `MessageTriggerAction.trigger` as a `Consumer<Message<?>>` method reference
and use a `LambdaMessageProcessor` for direct call
* Add a `Consumer` support for `LambdaMessageProcessor`

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

@@ -187,8 +187,8 @@ protected MessageChannel getCurrentMessageChannel() {
 	 */
 	protected InterceptableChannel currentInterceptableChannel() {
 		MessageChannel currentChannel = getCurrentMessageChannel();
-		if (currentChannel instanceof InterceptableChannel) {
-			return (InterceptableChannel) currentChannel;
+		if (currentChannel instanceof InterceptableChannel interceptableChannel) {
+			return interceptableChannel;
 		}
 		else {
 			DirectChannel newCurrentChannel = new DirectChannel();
@@ -1161,7 +1161,7 @@ public B bridge() {
 	 * @see GenericEndpointSpec
 	 */
 	public B bridge(Consumer<GenericEndpointSpec<BridgeHandler>> endpointConfigurer) {
-		return register(new GenericEndpointSpec<>(new BridgeHandler()), endpointConfigurer);
+		return handle(new BridgeHandler(), endpointConfigurer);
 	}
 
 	/**
@@ -2836,7 +2836,9 @@ public B trigger(MessageTriggerAction triggerAction) {
 	public B trigger(MessageTriggerAction triggerAction,
 			Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
 
-		return handle(new ServiceActivatingHandler(triggerAction, "trigger"), endpointConfigurer);
+		Consumer<Message<?>> trigger = triggerAction::trigger;
+		return handle(new ServiceActivatingHandler(new LambdaMessageProcessor(trigger, Message.class)),
+				endpointConfigurer);
 	}
 
 	/**

spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.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.
@@ -21,6 +21,7 @@
 import java.lang.reflect.Modifier;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.Consumer;
 import java.util.function.Function;
 
 import org.apache.commons.logging.Log;
@@ -98,7 +99,8 @@ public LambdaMessageProcessor(Object target, @Nullable Class<?> expectedType) {
 	}
 
 	private static boolean isExplicit(Object target) {
-		return target instanceof Function<?, ?> ||
+		return target instanceof Consumer<?> ||
+				target instanceof Function<?, ?> ||
 				target instanceof GenericHandler<?> ||
 				target instanceof GenericSelector<?> ||
 				target instanceof GenericTransformer<?, ?>;
@@ -189,7 +191,11 @@ else if (Map.class.isAssignableFrom(parameterType)) {
 
 	@SuppressWarnings({"unchecked", "rawtypes"})
 	private Object invokeMethod(Object[] args) throws InvocationTargetException, IllegalAccessException {
-		if (this.target instanceof Function function) {
+		if (this.target instanceof Consumer consumer) {
+			consumer.accept(args[0]);
+			return null;
+		}
+		else if (this.target instanceof Function function) {
 			return function.apply(args[0]);
 		}
 		else if (this.target instanceof GenericSelector selector) {