01d22edfd1
* Fix typos and language in README * Fix new line in the end of `application.yml` * Move `@Bean` for `RabbitMQContainer` as a regular `static` property on the class with a `@Container` * Use `@Testcontainers(disabledWithoutDocker = true)` * Move `@RabbitListener` method directly to test class * In the end we don't need extra `@TestConfiguration` class at all
60 lines
3.7 KiB
Plaintext
60 lines
3.7 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|flattenFunction`.
|
|
(The `flattenFunction` will be explained latter).
|
|
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.
|
|
|
|
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.
|
|
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. |