Fix Kotlin DSL according to IDEA errors and suggestions (#10307)

* Fix Kotlin DSL according to IDEA errors and suggestions

It is not clear why it does not fail with Gradle, but right now it is not possible to build the project using IDEA compiler.

Some of the suggestions are good and have led to the Nullability fixes in related Java DSL classes.
The `Any` extension requirement for the generic argument of reified types is not clear, but works.
The requirement to refine nullable Kotlin types is good.

Somewhat related to: #10083

* * Address more spotted Nullability concerns

* More Nullability refinement in the `BaseIntegrationFlowDefinition`

Author: artembilan

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

@@ -650,7 +650,7 @@ public <P> B convert(Class<P> payloadType) {
 	 * @see MethodInvokingTransformer
 	 * @see LambdaMessageProcessor
 	 */
-	public <P, T> B transform(@Nullable Class<P> expectedType, GenericTransformer<P, T> genericTransformer) {
+	public <P, T> B transform(Class<P> expectedType, GenericTransformer<P, T> genericTransformer) {
 		return transformWith((transformerSpec) ->
 				transformerSpec.transformer(genericTransformer).expectedType(expectedType));
 	}
@@ -737,7 +737,7 @@ public B filter(String expression) {
 	 */
 	public B filter(String expression, @Nullable Consumer<FilterEndpointSpec> endpointConfigurer) {
 		Assert.hasText(expression, "'expression' must not be empty");
-		return filter(null, new ExpressionEvaluatingSelector(expression), endpointConfigurer);
+		return doFilter(null, new ExpressionEvaluatingSelector(expression), endpointConfigurer);
 	}
 
 	/**
@@ -779,7 +779,7 @@ public B filter(Object service, @Nullable String methodName,
 				StringUtils.hasText(methodName)
 						? new MethodInvokingSelector(service, methodName)
 						: new MethodInvokingSelector(service);
-		return filter(null, selector, endpointConfigurer);
+		return doFilter(null, selector, endpointConfigurer);
 	}
 
 	/**
@@ -819,7 +819,7 @@ public B filter(MessageProcessorSpec<?> messageProcessorSpec,
 		Assert.notNull(messageProcessorSpec, MESSAGE_PROCESSOR_SPEC_MUST_NOT_BE_NULL);
 		MessageProcessor<?> processor = messageProcessorSpec.getObject();
 		return addComponent(processor)
-				.filter(null, new MethodInvokingSelector(processor), endpointConfigurer);
+				.doFilter(null, new MethodInvokingSelector(processor), endpointConfigurer);
 	}
 
 	/**
@@ -839,7 +839,7 @@ public B filter(MessageProcessorSpec<?> messageProcessorSpec,
 	 * @return the current {@link BaseIntegrationFlowDefinition}.
 	 * @see LambdaMessageProcessor
 	 */
-	public <P> B filter(@Nullable Class<P> expectedType, GenericSelector<P> genericSelector) {
+	public <P> B filter(Class<P> expectedType, GenericSelector<P> genericSelector) {
 		return filter(expectedType, genericSelector, null);
 	}
 
@@ -863,7 +863,13 @@ public <P> B filter(@Nullable Class<P> expectedType, GenericSelector<P> genericS
 	 * @see LambdaMessageProcessor
 	 * @see FilterEndpointSpec
 	 */
-	public <P> B filter(@Nullable Class<P> expectedType, GenericSelector<P> genericSelector,
+	public <P> B filter(Class<P> expectedType, GenericSelector<P> genericSelector,
+			@Nullable Consumer<FilterEndpointSpec> endpointConfigurer) {
+
+		return doFilter(expectedType, genericSelector, endpointConfigurer);
+	}
+
+	protected <P> B doFilter(@Nullable Class<P> expectedType, GenericSelector<P> genericSelector,
 			@Nullable Consumer<FilterEndpointSpec> endpointConfigurer) {
 
 		Assert.notNull(genericSelector, "'genericSelector' must not be null");
@@ -1002,7 +1008,7 @@ public B handle(Object service, @Nullable String methodName,
 	 * @return the current {@link BaseIntegrationFlowDefinition}.
 	 * @see LambdaMessageProcessor
 	 */
-	public <P> B handle(@Nullable Class<P> expectedType, GenericHandler<P> handler) {
+	public <P> B handle(Class<P> expectedType, GenericHandler<P> handler) {
 		return handle(expectedType, handler, null);
 	}
 
@@ -1026,7 +1032,13 @@ public <P> B handle(@Nullable Class<P> expectedType, GenericHandler<P> handler)
 	 * @return the current {@link BaseIntegrationFlowDefinition}.
 	 * @see LambdaMessageProcessor
 	 */
-	public <P> B handle(@Nullable Class<P> expectedType, GenericHandler<P> handler,
+	public <P> B handle(Class<P> expectedType, GenericHandler<P> handler,
+			@Nullable Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
+
+		return doHandle(expectedType, handler, endpointConfigurer);
+	}
+
+	protected  <P> B doHandle(@Nullable Class<P> expectedType, GenericHandler<P> handler,
 			@Nullable Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
 
 		ServiceActivatingHandler serviceActivatingHandler;
@@ -1558,7 +1570,7 @@ public B claimCheckOut(MessageStore messageStore, boolean removeMessage,
 	 * @return the current {@link BaseIntegrationFlowDefinition}.
 	 */
 	public B resequence() {
-		return resequence(null);
+		return register(new ResequencerSpec(), null);
 	}
 
 	/**
@@ -1579,7 +1591,7 @@ public B resequence() {
 	 * @return the current {@link BaseIntegrationFlowDefinition}.
 	 * @see ResequencerSpec
 	 */
-	public B resequence(@Nullable Consumer<ResequencerSpec> resequencer) {
+	public B resequence(Consumer<ResequencerSpec> resequencer) {
 		return register(new ResequencerSpec(), resequencer);
 	}
 
@@ -1588,7 +1600,7 @@ public B resequence(@Nullable Consumer<ResequencerSpec> resequencer) {
 	 * @return the current {@link BaseIntegrationFlowDefinition}.
 	 */
 	public B aggregate() {
-		return aggregate(null);
+		return register(new AggregatorSpec(), null);
 	}
 
 	/**
@@ -1617,7 +1629,7 @@ public B aggregate(Object aggregatorProcessor) {
 	 * @return the current {@link BaseIntegrationFlowDefinition}.
 	 * @see AggregatorSpec
 	 */
-	public B aggregate(@Nullable Consumer<AggregatorSpec> aggregator) {
+	public B aggregate(Consumer<AggregatorSpec> aggregator) {
 		return register(new AggregatorSpec(), aggregator);
 	}
 
@@ -1641,7 +1653,7 @@ public B route(String beanName, @Nullable String method) {
 	 * @return the current {@link BaseIntegrationFlowDefinition}.
 	 */
 	public B route(String beanName, @Nullable String method,
-			@Nullable Consumer<RouterSpec<Object, MethodInvokingRouter>> routerConfigurer) {
+			@Nullable Consumer<RouterSpec<@Nullable Object, MethodInvokingRouter>> routerConfigurer) {
 
 		MethodInvokingRouter methodInvokingRouter =
 				new MethodInvokingRouter(new BeanNameMessageProcessor<>(beanName, method));
@@ -1681,7 +1693,7 @@ public B route(Object service, @Nullable String methodName) {
 	 * @see MethodInvokingRouter
 	 */
 	public B route(Object service, @Nullable String methodName,
-			@Nullable Consumer<RouterSpec<Object, MethodInvokingRouter>> routerConfigurer) {
+			@Nullable Consumer<RouterSpec<@Nullable Object, MethodInvokingRouter>> routerConfigurer) {
 
 		MethodInvokingRouter router;
 		if (StringUtils.hasText(methodName)) {
@@ -1711,7 +1723,7 @@ public B route(String expression) {
 	 * @param <T> the target result type.
 	 * @return the current {@link BaseIntegrationFlowDefinition}.
 	 */
-	public <T> B route(String expression,
+	public <T extends @Nullable Object> B route(String expression,
 			@Nullable Consumer<RouterSpec<T, ExpressionEvaluatingRouter>> routerConfigurer) {
 
 		return route(new RouterSpec<>(new ExpressionEvaluatingRouter(PARSER.parseExpression(expression))),
@@ -1736,7 +1748,7 @@ public <T> B route(String expression,
 	 * @return the current {@link BaseIntegrationFlowDefinition}.
 	 * @see LambdaMessageProcessor
 	 */
-	public <S, T> B route(@Nullable Class<S> expectedType, Function<S, T> router) {
+	public <S, T extends @Nullable Object> B route(Class<S> expectedType, Function<S, T> router) {
 		return route(expectedType, router, null);
 	}
 
@@ -1764,7 +1776,13 @@ public <S, T> B route(@Nullable Class<S> expectedType, Function<S, T> router) {
 	 * @return the current {@link BaseIntegrationFlowDefinition}.
 	 * @see LambdaMessageProcessor
 	 */
-	public <P, T> B route(@Nullable Class<P> expectedType, Function<P, T> router,
+	public <P, T extends @Nullable Object> B route(Class<P> expectedType, Function<P, T> router,
+			@Nullable Consumer<RouterSpec<T, MethodInvokingRouter>> routerConfigurer) {
+
+		return doRoute(expectedType, router, routerConfigurer);
+	}
+
+	protected <P, T extends @Nullable Object> B doRoute(@Nullable Class<P> expectedType, Function<P, T> router,
 			@Nullable Consumer<RouterSpec<T, MethodInvokingRouter>> routerConfigurer) {
 
 		MethodInvokingRouter methodInvokingRouter =
@@ -1787,7 +1805,7 @@ public <P, T> B route(@Nullable Class<P> expectedType, Function<P, T> router,
 	 * @return the current {@link BaseIntegrationFlowDefinition}.
 	 */
 	public B route(MessageProcessorSpec<?> messageProcessorSpec) {
-		return route(messageProcessorSpec, (Consumer<RouterSpec<Object, MethodInvokingRouter>>) null);
+		return route(messageProcessorSpec, (Consumer<RouterSpec<@Nullable Object, MethodInvokingRouter>>) null);
 	}
 
 	/**
@@ -1807,7 +1825,7 @@ public B route(MessageProcessorSpec<?> messageProcessorSpec) {
 	 * @return the current {@link BaseIntegrationFlowDefinition}.
 	 */
 	public B route(MessageProcessorSpec<?> messageProcessorSpec,
-			@Nullable Consumer<RouterSpec<Object, MethodInvokingRouter>> routerConfigurer) {
+			@Nullable Consumer<RouterSpec<@Nullable Object, MethodInvokingRouter>> routerConfigurer) {
 
 		Assert.notNull(messageProcessorSpec, MESSAGE_PROCESSOR_SPEC_MUST_NOT_BE_NULL);
 		MessageProcessor<?> processor = messageProcessorSpec.getObject();

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

@@ -19,6 +19,8 @@
 import java.util.function.Consumer;
 import java.util.function.Function;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.integration.core.GenericHandler;
 import org.springframework.integration.core.GenericSelector;
 import org.springframework.integration.core.GenericTransformer;
@@ -83,7 +85,7 @@ public <S, T> B transform(GenericTransformer<S, T> genericTransformer) {
 	 * @return the current {@link IntegrationFlowDefinition}.
 	 */
 	public <P> B filter(GenericSelector<P> genericSelector) {
-		return filter(null, genericSelector);
+		return doFilter(null, genericSelector, null);
 	}
 
 	/**
@@ -106,7 +108,7 @@ public <P> B filter(GenericSelector<P> genericSelector) {
 	 * @see FilterEndpointSpec
 	 */
 	public <P> B filter(GenericSelector<P> genericSelector, Consumer<FilterEndpointSpec> endpointConfigurer) {
-		return filter(null, genericSelector, endpointConfigurer);
+		return doFilter(null, genericSelector, endpointConfigurer);
 	}
 
 	/**
@@ -127,7 +129,7 @@ public <P> B filter(GenericSelector<P> genericSelector, Consumer<FilterEndpointS
 	 * @see org.springframework.integration.handler.LambdaMessageProcessor
 	 */
 	public <P> B handle(GenericHandler<P> handler) {
-		return handle(null, handler);
+		return doHandle(null, handler, null);
 	}
 
 	/**
@@ -153,7 +155,7 @@ public <P> B handle(GenericHandler<P> handler) {
 	public <P> B handle(GenericHandler<P> handler,
 			Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
 
-		return handle(null, handler, endpointConfigurer);
+		return doHandle(null, handler, endpointConfigurer);
 	}
 
 	/**
@@ -171,8 +173,8 @@ public <P> B handle(GenericHandler<P> handler,
 	 * @param <T> the target result type.
 	 * @return the current {@link IntegrationFlowDefinition}.
 	 */
-	public <S, T> B route(Function<S, T> router) {
-		return route(null, router);
+	public <S, T extends @Nullable Object> B route(Function<S, T> router) {
+		return doRoute(null, router,  null);
 	}
 
 	/**
@@ -196,8 +198,10 @@ public <S, T> B route(Function<S, T> router) {
 	 * @param <T> the target result type.
 	 * @return the current {@link IntegrationFlowDefinition}.
 	 */
-	public <S, T> B route(Function<S, T> router, Consumer<RouterSpec<T, MethodInvokingRouter>> routerConfigurer) {
-		return route(null, router, routerConfigurer);
+	public <S, T extends @Nullable Object> B route(Function<S, T> router,
+			Consumer<RouterSpec<T, MethodInvokingRouter>> routerConfigurer) {
+
+		return doRoute(null, router, routerConfigurer);
 	}
 
 }

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

@@ -43,7 +43,8 @@
  *
  * @since 5.0
  */
-public class RouterSpec<K, R extends AbstractMappingMessageRouter> extends AbstractRouterSpec<RouterSpec<K, R>, R> {
+public class RouterSpec<K extends @Nullable Object, R extends AbstractMappingMessageRouter>
+		extends AbstractRouterSpec<RouterSpec<K, R>, R> {
 
 	private final RouterMappingProvider mappingProvider;
 

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

@@ -161,7 +161,7 @@ public <P> SplitterSpec function(Function<P, ?> function) {
 	 * @param expectedType the {@link Function} input argument type.
 	 * @return the spec.
 	 */
-	public SplitterSpec expectedType(@Nullable Class<?> expectedType) {
+	public SplitterSpec expectedType(Class<?> expectedType) {
 		this.expectedType = expectedType;
 		return this;
 	}

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

@@ -140,7 +140,7 @@ public <P, T> TransformerEndpointSpec transformer(GenericTransformer<P, T> trans
 	 * @param <P> the type ot expect.
 	 * @return the spec.
 	 */
-	public <P> TransformerEndpointSpec expectedType(@Nullable Class<P> expectedType) {
+	public <P> TransformerEndpointSpec expectedType(Class<P> expectedType) {
 		this.expectedType = expectedType;
 		return this;
 	}

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

@@ -67,11 +67,11 @@ abstract class KotlinConsumerEndpointSpec<S : ConsumerEndpointSpec<S, H>, H : Me
 		this.delegate.taskScheduler(taskScheduler)
 	}
 
-	fun handleMessageAdvice(vararg interceptors: MethodInterceptor?) {
+	fun handleMessageAdvice(vararg interceptors: MethodInterceptor) {
 		this.delegate.handleMessageAdvice(*interceptors)
 	}
 
-	fun advice(vararg advice: Advice?) {
+	fun advice(vararg advice: Advice) {
 		this.delegate.advice(*advice)
 	}
 
@@ -95,11 +95,11 @@ abstract class KotlinConsumerEndpointSpec<S : ConsumerEndpointSpec<S, H>, H : Me
 		this.delegate.transactional(handleMessageAdvice)
 	}
 
-	fun <T : Any?, V : Any?> customizeMonoReply(replyCustomizer: (Message<*>, Mono<T>) -> Publisher<V>) {
+	fun <T : Any, V : Any> customizeMonoReply(replyCustomizer: (Message<*>, Mono<T>) -> Publisher<V>) {
 		this.delegate.customizeMonoReply(replyCustomizer)
 	}
 
-	fun id(id: String?) {
+	fun id(id: String) {
 		this.delegate.id(id)
 	}