GH-107: Make Splitter Function as Flux-based

Fixes: #107

When we have a composition like this:

```
spring.cloud.function.definition = fileSupplier|splitterFunction
```

Then final "function" signature is like this `Supplier<Flux<Message<List<Message<?>>>>>`.
And that is exactly what we don't expected from the splitter in the end of the composition.
While Spring Cloud Stream supports de-batching, it works for a `List` output only if function is bound by itself.
In case of composition we got just a `Supplier`.

* Rework `SplitterFunctionConfiguration` for `splitterFunction` from `Function<Message<?>, List<Message<?>>>`
to `Function<Flux<Message<?>>, Flux<Message<?>>>` signature to support every possible simple and composed bindings
in Spring Cloud Stream
* Rework `SplitterFunctionApplicationTests` for new expected `Function<Flux<Message<?>>, Flux<Message<?>>>` signature
* Rework `zip-split-rabbit-binder` sample to not use a `flattenFunction` workaround
and fully rely on whatever is new for the `splitterFunction`
* Fix `ZipSplitRabbitBinderApplicationTests` moving the `@RabbitListener` into a `@TestConfiguration`.
Apparently in a new Spring Boot version the test class is registered as a bean much later than normal application context startup.
Therefore, even if the `@RabbitListener` parsed and registered properly, the `RabbitAdmin` bean
has been already started to see our extra bean definition for the `@QueueBinding`

Changing signature for the splitterFunction to reactive types would make it working even with a Supplier composition.

Fix JDBC & MongoDB suppliers to deal with a new version of Splitter function

Fix Checkstyle violations

Use `IntegrationReactiveUtils.messageSourceToFlux()` API

The `IntegrationReactiveUtils.messageSourceToFlux()` provides convenient API to represent a `MessageSource`
as a `Flux` to poll this source.
The API has an error handling logic and delay when no data emitted by the source

* Remove `org.springframework.cloud` dependencies from the project
since we don't use `@PollableBean` anymore, which comes from the `spring-cloud-function-context`
* Simplify `JdbcSupplierConfiguration` and `MongodbSupplierConfiguration` code more: more injections to the respective bean method.
* Use `(__) ->` lambda syntax for unused argument
* Remove unused `ThreadLocalFluxSinkMessageChannel` internal class
* Update Copyrights of the classes in this change

Upgrade to Gradle `8.12`

Author: artembilan

build.gradle

@@ -58,7 +58,6 @@ allprojects {
 			mavenBom "io.debezium:debezium-bom:$debeziumVersion"
 			mavenBom "io.awspring.cloud:spring-cloud-aws-dependencies:$springCloudAwsVersion"
 			mavenBom "org.springframework.boot:spring-boot-dependencies:$springBootVersion"
-			mavenBom "org.springframework.cloud:spring-cloud-dependencies:$springCloudVersion"
 			mavenBom "ai.djl:bom:$djlVersion"
 		}
 	}

dependencies.gradle

@@ -1,6 +1,5 @@
 ext {
     springBootVersion = '3.4.1'
-    springCloudVersion = '2024.0.0'
     springCloudAwsVersion = '3.2.1'
 
     debeziumVersion = '3.0.6.Final'

function/spring-splitter-function/src/main/java/org/springframework/cloud/fn/splitter/SplitterFunctionConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2024 the original author or authors.
+ * Copyright 2011-2025 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,11 +17,9 @@
 package org.springframework.cloud.fn.splitter;
 
 import java.nio.charset.Charset;
-import java.util.List;
 import java.util.Optional;
 import java.util.function.Function;
 
-import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 
 import org.springframework.beans.factory.annotation.Qualifier;
@@ -32,13 +30,12 @@
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Conditional;
-import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel;
+import org.springframework.integration.channel.FluxMessageChannel;
 import org.springframework.integration.file.splitter.FileSplitter;
 import org.springframework.integration.splitter.AbstractMessageSplitter;
 import org.springframework.integration.splitter.DefaultMessageSplitter;
 import org.springframework.integration.splitter.ExpressionEvaluatingSplitter;
 import org.springframework.messaging.Message;
-import org.springframework.messaging.MessageChannel;
 
 /**
  * Auto-configuration for Splitter function.
@@ -51,7 +48,7 @@
 public class SplitterFunctionConfiguration {
 
 	@Bean
-	public Function<Message<?>, List<Message<?>>> splitterFunction(
+	public Function<Flux<Message<?>>, Flux<Message<?>>> splitterFunction(
 			@Qualifier("expressionSplitter") Optional<AbstractMessageSplitter> expressionSplitter,
 			@Qualifier("fileSplitter") Optional<AbstractMessageSplitter> fileSplitter,
 			@Qualifier("defaultSplitter") Optional<AbstractMessageSplitter> defaultSplitter,
@@ -60,13 +57,13 @@ public Function<Message<?>, List<Message<?>>> splitterFunction(
 		AbstractMessageSplitter messageSplitter = expressionSplitter.or(() -> fileSplitter)
 			.or(() -> defaultSplitter)
 			.get();
+
 		messageSplitter.setApplySequence(splitterFunctionProperties.isApplySequence());
-		ThreadLocalFluxSinkMessageChannel outputChannel = new ThreadLocalFluxSinkMessageChannel();
+		FluxMessageChannel inputChannel = new FluxMessageChannel();
+		inputChannel.subscribe(messageSplitter);
+		FluxMessageChannel outputChannel = new FluxMessageChannel();
 		messageSplitter.setOutputChannel(outputChannel);
-		return (message) -> {
-			messageSplitter.handleMessage(message);
-			return outputChannel.publisherThreadLocal.get();
-		};
+		return (messageFlux) -> Flux.from(outputChannel).doOnRequest((__) -> inputChannel.subscribeTo(messageFlux));
 	}
 
 	@Bean
@@ -117,22 +114,4 @@ static class FileMarkers {
 
 	}
 
-	private static final class ThreadLocalFluxSinkMessageChannel
-			implements MessageChannel, ReactiveStreamsSubscribableChannel {
-
-		private final ThreadLocal<List<Message<?>>> publisherThreadLocal = new ThreadLocal<>();
-
-		@Override
-		@SuppressWarnings("unchecked")
-		public void subscribeTo(Publisher<? extends Message<?>> publisher) {
-			this.publisherThreadLocal.set(Flux.from(publisher).collectList().cast(List.class).block());
-		}
-
-		@Override
-		public boolean send(Message<?> message, long l) {
-			throw new UnsupportedOperationException("This channel only supports a reactive 'subscribeTo()' ");
-		}
-
-	}
-
 }

function/spring-splitter-function/src/test/java/org/springframework/cloud/fn/splitter/SplitterFunctionApplicationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2024 the original author or authors.
+ * Copyright 2011-2025 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.
@@ -16,10 +16,12 @@
 
 package org.springframework.cloud.fn.splitter;
 
-import java.util.List;
+import java.time.Duration;
 import java.util.function.Function;
 
 import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
@@ -28,19 +30,18 @@
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.annotation.DirtiesContext;
 
-import static org.assertj.core.api.Assertions.assertThat;
-
 @SpringBootTest(properties = "splitter.expression=payload.split(',')")
 @DirtiesContext
 public class SplitterFunctionApplicationTests {
 
 	@Autowired
-	Function<Message<?>, List<Message<?>>> splitter;
+	Function<Flux<Message<?>>, Flux<Message<?>>> splitter;
 
 	@Test
 	public void testExpressionSplitter() {
-		List<Message<?>> messageList = this.splitter.apply(new GenericMessage<>("hello,world"));
-		assertThat(messageList).extracting((m) -> m.getPayload().toString()).contains("hello", "world");
+		Flux<Message<?>> messageFlux = this.splitter.apply(Flux.just(new GenericMessage<>("hello,world")));
+		Flux<String> payloads = messageFlux.map(Message::getPayload).map(Object::toString);
+		StepVerifier.create(payloads).expectNext("hello", "world").thenCancel().verify(Duration.ofSeconds(30));
 	}
 
 	@SpringBootApplication

gradle/wrapper/gradle-wrapper.jar

@@ -1,7 +1,7 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionSha256Sum=544c35d6bd849ae8a5ed0bcea39ba677dc40f49df7d1835561582da2009b961d
-distributionUrl=https\://services.gradle.org/distributions/gradle-8.7-bin.zip
+distributionSha256Sum=7a00d51fb93147819aab76024feece20b6b84e420694101f276be952e08bef03
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.12-bin.zip
 networkTimeout=10000
 validateDistributionUrl=true
 zipStoreBase=GRADLE_USER_HOME

gradle/wrapper/gradle-wrapper.properties

@@ -15,6 +15,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+# SPDX-License-Identifier: Apache-2.0
+#
 
 ##############################################################################
 #
@@ -55,7 +57,7 @@
 #       Darwin, MinGW, and NonStop.
 #
 #   (3) This script is generated from the Groovy template
-#       https://github.com/gradle/gradle/blob/HEAD/subprojects/plugins/src/main/resources/org/gradle/api/internal/plugins/unixStartScript.txt
+#       https://github.com/gradle/gradle/blob/HEAD/platforms/jvm/plugins-application/src/main/resources/org/gradle/api/internal/plugins/unixStartScript.txt
 #       within the Gradle project.
 #
 #       You can find Gradle at https://github.com/gradle/gradle/.
@@ -84,7 +86,7 @@ done
 # shellcheck disable=SC2034
 APP_BASE_NAME=${0##*/}
 # Discard cd standard output in case $CDPATH is set (https://github.com/gradle/gradle/issues/25036)
-APP_HOME=$( cd "${APP_HOME:-./}" > /dev/null && pwd -P ) || exit
+APP_HOME=$( cd -P "${APP_HOME:-./}" > /dev/null && printf '%s\n' "$PWD" ) || exit
 
 # Use the maximum available, or set MAX_FD != -1 to use that value.
 MAX_FD=maximum

gradlew

@@ -13,6 +13,8 @@
 @rem See the License for the specific language governing permissions and
 @rem limitations under the License.
 @rem
+@rem SPDX-License-Identifier: Apache-2.0
+@rem
 
 @if "%DEBUG%"=="" @echo off
 @rem ##########################################################################

gradlew.bat

@@ -18,8 +18,7 @@ The second one is for `UnZipTransformer`, which we use for a custom function to
 The `splitterFunction` is used in a `FileSplitter` mode to read lines from unzipped entries and emit each of them as an individual message.
 Essentially, we are splitting twice: zip entries, and content of each file.
 
-The composition is like this: `fileSupplier|unzipFunction|splitterFunction|flattenFunction`.
-(The `flattenFunction` will be explained latter).
+The composition is like this: `fileSupplier|unzipFunction|splitterFunction.
 The result of this composition is a `Supplier<Flux<Mesage<?>>>` and we bind it into a RabbitMQ `unzipped_data_exchange` using Spring Cloud Stream.
 
 For `fileSupplier` we provide these configuration properties:
@@ -49,8 +48,6 @@ Which is a trigger for that function to use a `FileSplitter` for zip entries to
 The custom `ZipSplitRabbitBinderApplication.unzipFunction()` (might be a candidate for the future Functions Catalog version) uses `Flux` API to unzip polled files via `UnZipTransformer` and then `flatMapIterable()` for zip entries.
 Then those entries are fed into a `splitterFunction` for `FileSplitter` mode.
 
-The mention `ZipSplitRabbitBinderApplication.flattenFunction()` is needed for now here since `splitterFucntion` produces a `List<Message>` which cannot be https://docs.spring.io/spring-cloud-stream/reference/spring-cloud-stream/producing-and-consuming-messages.html#batch-producers[de-batched] by Spring Cloud Stream since our final product of the composition is, essentially, `Supplier<Flux<Message<?>>>`.
-
 To run the application from main `ZipSplitRabbitBinderApplication` class (`./gradlew bootRun`), the RabbitMQ broker must be supplied on the target environment.
 
 The test environment for this sample uses `org.springframework.boot:spring-boot-testcontainers` and `org.testcontainers:rabbitmq` to run RabbitMQ in Docker container and wire it properly into Spring Boot auto-configuration.

samples/zip-split-rabbit-binder/README.adoc

@@ -1,7 +1,6 @@
 package com.example;
 
 import java.io.File;
-import java.util.List;
 import java.util.Map;
 import java.util.function.Function;
 
@@ -34,10 +33,4 @@ Function<Flux<Message<File>>, Flux<File>> unzipFunction(UnZipTransformer unZipTr
 			.flatMapIterable(Map::values);
 	}
 
-	// TODO until 'splitterFunction' is fixed this way: https://github.com/spring-cloud/spring-functions-catalog/issues/107
-	@Bean
-	Function<Flux<Message<List<Message<?>>>>, Flux<Message<?>>> flattenFunction() {
-		return messageFlux -> messageFlux.map(Message::getPayload).flatMapIterable(Function.identity());
-	}
-
 }

samples/zip-split-rabbit-binder/src/main/java/com/example/ZipSplitRabbitBinderApplication.java

@@ -4,11 +4,11 @@ spring:
 
   cloud:
     function:
-      definition: fileSupplier|unzipFunction|splitterFunction|flattenFunction
+      definition: fileSupplier|unzipFunction|splitterFunction
 
     stream:
       bindings:
-        fileSupplier|unzipFunction|splitterFunction|flattenFunction-out-0:
+        fileSupplier|unzipFunction|splitterFunction-out-0:
           destination: unzipped_data_exchange
 
 file:

samples/zip-split-rabbit-binder/src/main/resources/application.yml

@@ -19,6 +19,7 @@
 import org.springframework.amqp.rabbit.annotation.QueueBinding;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
 import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.context.TestConfiguration;
 import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
 import org.springframework.test.annotation.DirtiesContext;
 
@@ -47,11 +48,16 @@ void zippedFilesAreSplittedToRabbitBinding() throws InterruptedException {
 		}
 	}
 
-	@RabbitListener(bindings = @QueueBinding(value = @Queue,
-			exchange = @Exchange(value = "unzipped_data_exchange", type = ExchangeTypes.TOPIC), key = "#"))
-	void receiveDataFromSplittedZips(String payload) {
-		LOG.info("A line from zip entry: " + payload);
-		DATA_SINK.offer(payload);
+	@TestConfiguration
+	static class RabbitListenerTestConfiguration {
+
+		@RabbitListener(bindings = @QueueBinding(value = @Queue,
+				exchange = @Exchange(value = "unzipped_data_exchange", type = ExchangeTypes.TOPIC), key = "#"))
+		void receiveDataFromSplittedZips(String payload) {
+			LOG.info("A line from zip entry: " + payload);
+			DATA_SINK.offer(payload);
+		}
+
 	}
 
 }

samples/zip-split-rabbit-binder/src/test/java/com/example/ZipSplitRabbitBinderApplicationTests.java

@@ -2,7 +2,6 @@ dependencies {
     api project(':spring-splitter-function')
     api 'org.springframework.integration:spring-integration-jdbc'
     api 'org.springframework.boot:spring-boot-starter-jdbc'
-    api 'org.springframework.cloud:spring-cloud-function-context'
 
     runtimeOnly 'org.hsqldb:hsqldb'
     runtimeOnly 'com.h2database:h2'

supplier/spring-jdbc-supplier/build.gradle

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2024 the original author or authors.
+ * Copyright 2019-2025 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.
@@ -16,7 +16,6 @@
 
 package org.springframework.cloud.fn.supplier.jdbc;
 
-import java.util.List;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
@@ -30,10 +29,10 @@
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.fn.common.config.ComponentCustomizer;
 import org.springframework.cloud.fn.splitter.SplitterFunctionConfiguration;
-import org.springframework.cloud.function.context.PollableBean;
 import org.springframework.context.annotation.Bean;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.jdbc.JdbcPollingChannelAdapter;
+import org.springframework.integration.util.IntegrationReactiveUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 
@@ -47,45 +46,26 @@
 @EnableConfigurationProperties(JdbcSupplierProperties.class)
 public class JdbcSupplierConfiguration {
 
-	private final JdbcSupplierProperties properties;
-
-	private final DataSource dataSource;
-
-	public JdbcSupplierConfiguration(JdbcSupplierProperties properties, DataSource dataSource) {
-		this.properties = properties;
-		this.dataSource = dataSource;
-	}
-
 	@Bean
-	public MessageSource<Object> jdbcMessageSource(
+	public JdbcPollingChannelAdapter jdbcMessageSource(JdbcSupplierProperties properties, DataSource dataSource,
 			@Nullable ComponentCustomizer<JdbcPollingChannelAdapter> jdbcPollingChannelAdapterCustomizer) {
 
-		JdbcPollingChannelAdapter jdbcPollingChannelAdapter = new JdbcPollingChannelAdapter(this.dataSource,
-				this.properties.getQuery());
-		jdbcPollingChannelAdapter.setMaxRows(this.properties.getMaxRows());
-		jdbcPollingChannelAdapter.setUpdateSql(this.properties.getUpdate());
+		JdbcPollingChannelAdapter jdbcPollingChannelAdapter = new JdbcPollingChannelAdapter(dataSource,
+				properties.getQuery());
+		jdbcPollingChannelAdapter.setMaxRows(properties.getMaxRows());
+		jdbcPollingChannelAdapter.setUpdateSql(properties.getUpdate());
 		if (jdbcPollingChannelAdapterCustomizer != null) {
 			jdbcPollingChannelAdapterCustomizer.customize(jdbcPollingChannelAdapter);
 		}
 		return jdbcPollingChannelAdapter;
 	}
 
 	@Bean(name = "jdbcSupplier")
-	@PollableBean
 	@ConditionalOnProperty(prefix = "jdbc.supplier", name = "split", matchIfMissing = true)
-	public Supplier<Flux<Message<?>>> splittedSupplier(MessageSource<Object> jdbcMessageSource,
-			Function<Message<?>, List<Message<?>>> splitterFunction) {
+	public Supplier<Flux<Message<?>>> splittedSupplier(JdbcPollingChannelAdapter jdbcMessageSource,
+			Function<Flux<Message<Object>>, Flux<Message<?>>> splitterFunction) {
 
-		return () -> {
-			Message<?> received = jdbcMessageSource.receive();
-			if (received != null) {
-				// multiple Message<Map<String, Object>>
-				return Flux.fromIterable(splitterFunction.apply(received));
-			}
-			else {
-				return Flux.empty();
-			}
-		};
+		return () -> IntegrationReactiveUtils.messageSourceToFlux(jdbcMessageSource).transform(splitterFunction);
 	}
 
 	@Bean