3ebce8858f
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`
57 lines
3.2 KiB
Plaintext
57 lines
3.2 KiB
Plaintext
= Split unzipped content into Rabbit Binder destination
|
|
|
|
This sample demonstrates how out of the box catalog functions and custom functions can be composed, and how the final result can be bound to RabbitMQ destination by the https://spring.io/projects/spring-cloud-function[Spring Cloud Stream] framework.
|
|
The goal of this sample is to poll zip files, unzip them, and emit messages to RabbitMQ for every line of text from those zip entries.
|
|
|
|
The sample uses these dependencies:
|
|
|
|
[source,groovy]
|
|
----
|
|
implementation 'org.springframework.cloud:spring-cloud-stream-binder-rabbit'
|
|
implementation 'org.springframework.integration:spring-integration-zip'
|
|
implementation 'org.springframework.cloud.fn:spring-file-supplier'
|
|
implementation 'org.springframework.cloud.fn:spring-splitter-function'
|
|
----
|
|
|
|
The first one is for binding output of the composed function into a RabbitMQ destination.
|
|
The second one is for `UnZipTransformer`, which we use for a custom function to unzip polled files by the `fileSupplier`.
|
|
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.
|
|
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:
|
|
|
|
[source,yaml]
|
|
----
|
|
file:
|
|
supplier:
|
|
directory: # Set some real dir with zips to process
|
|
filename-pattern: '*.zip'
|
|
----
|
|
|
|
Poll only zip files from the provided directory.
|
|
The sample doesn't come with one, so it's up to an end-user to provide specific directory with zip files.
|
|
Or this `fileSupplier` could be replaced with any other file-based supplier.
|
|
|
|
The `splitterFunction` comes with this property:
|
|
|
|
[source,yaml]
|
|
----
|
|
splitter:
|
|
charset: UTF-8
|
|
----
|
|
|
|
Which is a trigger for that function to use a `FileSplitter` for zip entries to emit their lines of text as individual messages.
|
|
|
|
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.
|
|
|
|
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.
|
|
The `ZipSplitRabbitBinderApplicationTests` uses `dirWithZips` directory from classpath with two zip files.
|
|
Then the `fileSupplier` polls those files and emits them into the mentioned function composition.
|
|
The `@RabbitListener` in the test configuration bind an anonymous queue to the mentioned `unzipped_data_exchange` topic exchange to consume produced data from our functions composition.
|
|
The test, by itself, verifies that all consumed by `@RabbitListener` data is, essentially, lines from the mentioned zipped files. |