diff --git a/build.gradle b/build.gradle index ed8ec9ee..8e9b11ad 100644 --- a/build.gradle +++ b/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" } } diff --git a/dependencies.gradle b/dependencies.gradle index 169132c3..4541910d 100644 --- a/dependencies.gradle +++ b/dependencies.gradle @@ -1,6 +1,5 @@ ext { springBootVersion = '3.4.1' - springCloudVersion = '2024.0.0' springCloudAwsVersion = '3.2.1' debeziumVersion = '3.0.6.Final' diff --git a/function/spring-splitter-function/src/main/java/org/springframework/cloud/fn/splitter/SplitterFunctionConfiguration.java b/function/spring-splitter-function/src/main/java/org/springframework/cloud/fn/splitter/SplitterFunctionConfiguration.java index ad1887aa..041452e7 100644 --- a/function/spring-splitter-function/src/main/java/org/springframework/cloud/fn/splitter/SplitterFunctionConfiguration.java +++ b/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.autoconfigure.condition.ConditionalOnProperty; 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 @@ import org.springframework.messaging.MessageChannel; public class SplitterFunctionConfiguration { @Bean - public Function, List>> splitterFunction( + public Function>, Flux>> splitterFunction( @Qualifier("expressionSplitter") Optional expressionSplitter, @Qualifier("fileSplitter") Optional fileSplitter, @Qualifier("defaultSplitter") Optional defaultSplitter, @@ -60,13 +57,13 @@ public class SplitterFunctionConfiguration { 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 @@ public class SplitterFunctionConfiguration { } - private static final class ThreadLocalFluxSinkMessageChannel - implements MessageChannel, ReactiveStreamsSubscribableChannel { - - private final ThreadLocal>> publisherThreadLocal = new ThreadLocal<>(); - - @Override - @SuppressWarnings("unchecked") - public void subscribeTo(Publisher> 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()' "); - } - - } - } diff --git a/function/spring-splitter-function/src/test/java/org/springframework/cloud/fn/splitter/SplitterFunctionApplicationTests.java b/function/spring-splitter-function/src/test/java/org/springframework/cloud/fn/splitter/SplitterFunctionApplicationTests.java index 910b1645..f0c0bafe 100644 --- a/function/spring-splitter-function/src/test/java/org/springframework/cloud/fn/splitter/SplitterFunctionApplicationTests.java +++ b/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.Message; 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, List>> splitter; + Function>, Flux>> splitter; @Test public void testExpressionSplitter() { - List> messageList = this.splitter.apply(new GenericMessage<>("hello,world")); - assertThat(messageList).extracting((m) -> m.getPayload().toString()).contains("hello", "world"); + Flux> messageFlux = this.splitter.apply(Flux.just(new GenericMessage<>("hello,world"))); + Flux payloads = messageFlux.map(Message::getPayload).map(Object::toString); + StepVerifier.create(payloads).expectNext("hello", "world").thenCancel().verify(Duration.ofSeconds(30)); } @SpringBootApplication diff --git a/gradle/wrapper/gradle-wrapper.jar b/gradle/wrapper/gradle-wrapper.jar index e6441136..a4b76b95 100644 Binary files a/gradle/wrapper/gradle-wrapper.jar and b/gradle/wrapper/gradle-wrapper.jar differ diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties index 381baa9c..e1b837a1 100644 --- a/gradle/wrapper/gradle-wrapper.properties +++ b/gradle/wrapper/gradle-wrapper.properties @@ -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 diff --git a/gradlew b/gradlew index 1aa94a42..f3b75f3b 100755 --- a/gradlew +++ b/gradlew @@ -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 diff --git a/gradlew.bat b/gradlew.bat index 25da30db..9d21a218 100644 --- a/gradlew.bat +++ b/gradlew.bat @@ -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 ########################################################################## diff --git a/samples/zip-split-rabbit-binder/README.adoc b/samples/zip-split-rabbit-binder/README.adoc index c552d792..da413ecb 100644 --- a/samples/zip-split-rabbit-binder/README.adoc +++ b/samples/zip-split-rabbit-binder/README.adoc @@ -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>>` 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` 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>>`. - 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. diff --git a/samples/zip-split-rabbit-binder/src/main/java/com/example/ZipSplitRabbitBinderApplication.java b/samples/zip-split-rabbit-binder/src/main/java/com/example/ZipSplitRabbitBinderApplication.java index 3a611906..49a3a3d3 100644 --- a/samples/zip-split-rabbit-binder/src/main/java/com/example/ZipSplitRabbitBinderApplication.java +++ b/samples/zip-split-rabbit-binder/src/main/java/com/example/ZipSplitRabbitBinderApplication.java @@ -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 @@ public class ZipSplitRabbitBinderApplication { .flatMapIterable(Map::values); } - // TODO until 'splitterFunction' is fixed this way: https://github.com/spring-cloud/spring-functions-catalog/issues/107 - @Bean - Function>>>, Flux>> flattenFunction() { - return messageFlux -> messageFlux.map(Message::getPayload).flatMapIterable(Function.identity()); - } - } diff --git a/samples/zip-split-rabbit-binder/src/main/resources/application.yml b/samples/zip-split-rabbit-binder/src/main/resources/application.yml index 904cd50c..cb535712 100644 --- a/samples/zip-split-rabbit-binder/src/main/resources/application.yml +++ b/samples/zip-split-rabbit-binder/src/main/resources/application.yml @@ -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: diff --git a/samples/zip-split-rabbit-binder/src/test/java/com/example/ZipSplitRabbitBinderApplicationTests.java b/samples/zip-split-rabbit-binder/src/test/java/com/example/ZipSplitRabbitBinderApplicationTests.java index 76ff0978..cb0a453d 100644 --- a/samples/zip-split-rabbit-binder/src/test/java/com/example/ZipSplitRabbitBinderApplicationTests.java +++ b/samples/zip-split-rabbit-binder/src/test/java/com/example/ZipSplitRabbitBinderApplicationTests.java @@ -19,6 +19,7 @@ import org.springframework.amqp.rabbit.annotation.Queue; 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 @@ class ZipSplitRabbitBinderApplicationTests { } } - @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); + } + } } diff --git a/supplier/spring-jdbc-supplier/build.gradle b/supplier/spring-jdbc-supplier/build.gradle index cc54e02b..71cbf393 100644 --- a/supplier/spring-jdbc-supplier/build.gradle +++ b/supplier/spring-jdbc-supplier/build.gradle @@ -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' diff --git a/supplier/spring-jdbc-supplier/src/main/java/org/springframework/cloud/fn/supplier/jdbc/JdbcSupplierConfiguration.java b/supplier/spring-jdbc-supplier/src/main/java/org/springframework/cloud/fn/supplier/jdbc/JdbcSupplierConfiguration.java index 3f157e10..49818908 100644 --- a/supplier/spring-jdbc-supplier/src/main/java/org/springframework/cloud/fn/supplier/jdbc/JdbcSupplierConfiguration.java +++ b/supplier/spring-jdbc-supplier/src/main/java/org/springframework/cloud/fn/supplier/jdbc/JdbcSupplierConfiguration.java @@ -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.autoconfigure.jdbc.DataSourceAutoConfiguration; 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,23 +46,14 @@ import org.springframework.messaging.Message; @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 jdbcMessageSource( + public JdbcPollingChannelAdapter jdbcMessageSource(JdbcSupplierProperties properties, DataSource dataSource, @Nullable ComponentCustomizer 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); } @@ -71,21 +61,11 @@ public class JdbcSupplierConfiguration { } @Bean(name = "jdbcSupplier") - @PollableBean @ConditionalOnProperty(prefix = "jdbc.supplier", name = "split", matchIfMissing = true) - public Supplier>> splittedSupplier(MessageSource jdbcMessageSource, - Function, List>> splitterFunction) { + public Supplier>> splittedSupplier(JdbcPollingChannelAdapter jdbcMessageSource, + Function>, Flux>> splitterFunction) { - return () -> { - Message received = jdbcMessageSource.receive(); - if (received != null) { - // multiple Message> - return Flux.fromIterable(splitterFunction.apply(received)); - } - else { - return Flux.empty(); - } - }; + return () -> IntegrationReactiveUtils.messageSourceToFlux(jdbcMessageSource).transform(splitterFunction); } @Bean diff --git a/supplier/spring-jdbc-supplier/src/test/java/org/springframework/cloud/fn/supplier/jdbc/DefaultJdbcSupplierTests.java b/supplier/spring-jdbc-supplier/src/test/java/org/springframework/cloud/fn/supplier/jdbc/DefaultJdbcSupplierTests.java index 87b29f1b..e40158ca 100644 --- a/supplier/spring-jdbc-supplier/src/test/java/org/springframework/cloud/fn/supplier/jdbc/DefaultJdbcSupplierTests.java +++ b/supplier/spring-jdbc-supplier/src/test/java/org/springframework/cloud/fn/supplier/jdbc/DefaultJdbcSupplierTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2024 the original author or authors. + * Copyright 2020-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. @@ -26,12 +26,15 @@ import reactor.test.StepVerifier; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.messaging.Message; import org.springframework.test.annotation.DirtiesContext; import static org.assertj.core.api.Assertions.assertThat; +/** + * @author Soby Chacko + * @author Artem Bilan + */ @SpringBootTest(properties = "jdbc.supplier.query=select id, name from test order by id") @DirtiesContext public class DefaultJdbcSupplierTests { @@ -39,14 +42,11 @@ public class DefaultJdbcSupplierTests { @Autowired Supplier>> jdbcSupplier; - @Autowired - JdbcTemplate jdbcTemplate; - @Test @SuppressWarnings("rawtypes") void testExtraction() { final Flux> messageFlux = jdbcSupplier.get(); - StepVerifier stepVerifier = StepVerifier.create(messageFlux) + StepVerifier.create(messageFlux) .assertNext((message) -> assertThat(message) .satisfies((msg) -> assertThat(msg).extracting(Message::getPayload).matches((o) -> { Map map = (Map) o; @@ -63,8 +63,7 @@ public class DefaultJdbcSupplierTests { return map.get("ID").equals(3L) && map.get("NAME").equals("John"); }))) .thenCancel() - .verifyLater(); - stepVerifier.verify(); + .verify(); } @SpringBootApplication diff --git a/supplier/spring-mongodb-supplier/build.gradle b/supplier/spring-mongodb-supplier/build.gradle index 0bed0ba5..b866abc0 100644 --- a/supplier/spring-mongodb-supplier/build.gradle +++ b/supplier/spring-mongodb-supplier/build.gradle @@ -2,7 +2,6 @@ dependencies { api project(':spring-splitter-function') api 'org.springframework.integration:spring-integration-mongodb' api 'org.mongodb:mongodb-driver-sync' - api 'org.springframework.cloud:spring-cloud-function-context' testImplementation 'org.testcontainers:mongodb' testImplementation project(':spring-mongodb-consumer').sourceSets.test.output diff --git a/supplier/spring-mongodb-supplier/src/main/java/org/springframework/cloud/fn/supplier/mongo/MongodbSupplierConfiguration.java b/supplier/spring-mongodb-supplier/src/main/java/org/springframework/cloud/fn/supplier/mongo/MongodbSupplierConfiguration.java index 693a461b..53c7d3a1 100644 --- a/supplier/spring-mongodb-supplier/src/main/java/org/springframework/cloud/fn/supplier/mongo/MongodbSupplierConfiguration.java +++ b/supplier/spring-mongodb-supplier/src/main/java/org/springframework/cloud/fn/supplier/mongo/MongodbSupplierConfiguration.java @@ -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.mongo; -import java.util.List; import java.util.function.Function; import java.util.function.Supplier; @@ -28,12 +27,12 @@ import org.springframework.boot.autoconfigure.mongo.MongoAutoConfiguration; 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.data.mongodb.core.MongoTemplate; import org.springframework.expression.Expression; import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.mongodb.inbound.MongoDbMessageSource; +import org.springframework.integration.util.IntegrationReactiveUtils; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; @@ -49,31 +48,12 @@ import org.springframework.messaging.Message; @EnableConfigurationProperties({ MongodbSupplierProperties.class }) public class MongodbSupplierConfiguration { - private final MongodbSupplierProperties properties; - - private final MongoTemplate mongoTemplate; - - public MongodbSupplierConfiguration(MongodbSupplierProperties properties, MongoTemplate mongoTemplate) { - this.properties = properties; - this.mongoTemplate = mongoTemplate; - } - @Bean(name = "mongodbSupplier") - @PollableBean @ConditionalOnProperty(prefix = "mongodb", name = "split", matchIfMissing = true) public Supplier>> splittedSupplier(MongoDbMessageSource mongoDbSource, - Function, List>> splitterFunction) { + Function>, Flux>> splitterFunction) { - return () -> { - Message received = mongoDbSource.receive(); - if (received != null) { - // multiple Message> - return Flux.fromIterable(splitterFunction.apply(received)); - } - else { - return Flux.empty(); - } - }; + return () -> IntegrationReactiveUtils.messageSourceToFlux(mongoDbSource).transform(splitterFunction); } @Bean @@ -83,15 +63,18 @@ public class MongodbSupplierConfiguration { } @Bean - public MongoDbMessageSource mongoDbSource( + public MongoDbMessageSource mongoDbSource(MongodbSupplierProperties properties, MongoTemplate mongoTemplate, @Nullable ComponentCustomizer mongoDbMessageSourceCustomizer) { - Expression queryExpression = (this.properties.getQueryExpression() != null) - ? this.properties.getQueryExpression() : new LiteralExpression(this.properties.getQuery()); - MongoDbMessageSource mongoDbMessageSource = new MongoDbMessageSource(this.mongoTemplate, queryExpression); - mongoDbMessageSource.setCollectionNameExpression(new LiteralExpression(this.properties.getCollection())); + Expression queryExpression = properties.getQueryExpression(); + if (queryExpression == null) { + queryExpression = new LiteralExpression(properties.getQuery()); + } + + MongoDbMessageSource mongoDbMessageSource = new MongoDbMessageSource(mongoTemplate, queryExpression); + mongoDbMessageSource.setCollectionNameExpression(new LiteralExpression(properties.getCollection())); mongoDbMessageSource.setEntityClass(String.class); - mongoDbMessageSource.setUpdateExpression(this.properties.getUpdateExpression()); + mongoDbMessageSource.setUpdateExpression(properties.getUpdateExpression()); if (mongoDbMessageSourceCustomizer != null) { mongoDbMessageSourceCustomizer.customize(mongoDbMessageSource); diff --git a/supplier/spring-mongodb-supplier/src/test/java/org/springframework/cloud/fn/supplier/mongo/MongodbSupplierApplicationTests.java b/supplier/spring-mongodb-supplier/src/test/java/org/springframework/cloud/fn/supplier/mongo/MongodbSupplierApplicationTests.java index efc98c56..0a9dc9be 100644 --- a/supplier/spring-mongodb-supplier/src/test/java/org/springframework/cloud/fn/supplier/mongo/MongodbSupplierApplicationTests.java +++ b/supplier/spring-mongodb-supplier/src/test/java/org/springframework/cloud/fn/supplier/mongo/MongodbSupplierApplicationTests.java @@ -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. @@ -76,8 +76,6 @@ class MongodbSupplierApplicationTests implements MongoDbTestContainerSupport { (message) -> assertThat(toMap(message)).contains(entry("greeting", "hola"), entry("name", "bar"))) .thenCancel() .verify(); - - assertThat(this.mongodbSupplier.get().collectList().block()).isEmpty(); } @SuppressWarnings("unchecked")