SpringCloud/spring-functions-catalog
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-107: Make Splitter Function as Flux-based

Browse Source 
Fixes: https://github.com/spring-cloud/spring-functions-catalog/issues/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`
This commit is contained in:
Artem Bilan
2025-01-16 17:15:41 -05:00
committed by GitHub
parent 9ce6a1b326
commit 3ebce8858f
18 changed files with 69 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
build.gradle
View File

@@ -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"
}
}

1
dependencies.gradle
View File

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

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

@@ -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<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 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<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()' ");
}
}
}

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

@@ -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<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

BIN
gradle/wrapper/gradle-wrapper.jar vendored
View File

Binary file not shown.

4
gradle/wrapper/gradle-wrapper.properties vendored
View File

@@ -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

6
gradlew vendored
View File

@@ -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

2
gradlew.bat vendored
View File

@@ -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 ##########################################################################

5
samples/zip-split-rabbit-binder/README.adoc
View File

@@ -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.

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

@@ -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<Message<List<Message<?>>>>, Flux<Message<?>>> flattenFunction() {
return messageFlux -> messageFlux.map(Message::getPayload).flatMapIterable(Function.identity());
}
}

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

@@ -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:

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

@@ -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);
}
}
}

1
supplier/spring-jdbc-supplier/build.gradle
View File

@@ -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'

40
supplier/spring-jdbc-supplier/src/main/java/org/springframework/cloud/fn/supplier/jdbc/JdbcSupplierConfiguration.java
View File

@@ -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<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);
}
@@ -71,21 +61,11 @@ public class JdbcSupplierConfiguration {
}
@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

15
supplier/spring-jdbc-supplier/src/test/java/org/springframework/cloud/fn/supplier/jdbc/DefaultJdbcSupplierTests.java
View File

@@ -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<Flux<Message<?>>> jdbcSupplier;
@Autowired
JdbcTemplate jdbcTemplate;
@Test
@SuppressWarnings("rawtypes")
void testExtraction() {
final Flux<Message<?>> 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

1
supplier/spring-mongodb-supplier/build.gradle
View File

@@ -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

43
supplier/spring-mongodb-supplier/src/main/java/org/springframework/cloud/fn/supplier/mongo/MongodbSupplierConfiguration.java
View File

@@ -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<Flux<Message<?>>> splittedSupplier(MongoDbMessageSource mongoDbSource,
Function<Message<?>, List<Message<?>>> splitterFunction) {
Function<Flux<Message<Object>>, Flux<Message<?>>> splitterFunction) {
return () -> {
Message<?> received = mongoDbSource.receive();
if (received != null) {
// multiple Message<Map<String, Object>>
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<MongoDbMessageSource> 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);

4
supplier/spring-mongodb-supplier/src/test/java/org/springframework/cloud/fn/supplier/mongo/MongodbSupplierApplicationTests.java
View File

@@ -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")