Address PR reviews:
* 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
This commit is contained in:
@@ -1,6 +1,6 @@
|
||||
= Split unzipped content into Rabbit Binder destination
|
||||
|
||||
This sample demonstrates how out of the box from catalog 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.
|
||||
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:
|
||||
@@ -20,7 +20,7 @@ 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).
|
||||
This becomes a `Supplier<Flux<Mesage<?>>>` and we bind it into a RabbitMQ `unzipped_data_exchange` using Spring Cloud Stream.
|
||||
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:
|
||||
|
||||
@@ -38,9 +38,10 @@ Or this `fileSupplier` could be replaced with any other file-based supplier.
|
||||
|
||||
The `splitterFunction` comes with this property:
|
||||
|
||||
[source,properties]
|
||||
[source,yaml]
|
||||
----
|
||||
splitter.charset=UTF-8
|
||||
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.
|
||||
|
||||
@@ -44,7 +44,6 @@ dependencies {
|
||||
testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
|
||||
}
|
||||
|
||||
|
||||
tasks.named('test') {
|
||||
useJUnitPlatform()
|
||||
}
|
||||
|
||||
@@ -17,4 +17,4 @@ file:
|
||||
filename-pattern: '*.zip'
|
||||
|
||||
splitter:
|
||||
charset: UTF-8
|
||||
charset: UTF-8
|
||||
|
||||
@@ -9,6 +9,8 @@ import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.testcontainers.containers.RabbitMQContainer;
|
||||
import org.testcontainers.junit.jupiter.Container;
|
||||
import org.testcontainers.junit.jupiter.Testcontainers;
|
||||
import org.testcontainers.utility.DockerImageName;
|
||||
|
||||
import org.springframework.amqp.core.ExchangeTypes;
|
||||
@@ -18,28 +20,22 @@ 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.testcontainers.service.connection.ServiceConnection;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Import;
|
||||
import org.springframework.core.io.ClassPathResource;
|
||||
import org.springframework.test.annotation.DirtiesContext;
|
||||
import org.springframework.test.context.DynamicPropertyRegistry;
|
||||
import org.springframework.test.context.DynamicPropertySource;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
@Import(ZipSplitRabbitBinderApplicationTests.TestConfiguration.class)
|
||||
@SpringBootTest
|
||||
@SpringBootTest(properties = "file.supplier.directory=classpath:/dirWithZips")
|
||||
@DirtiesContext
|
||||
@Testcontainers(disabledWithoutDocker = true)
|
||||
class ZipSplitRabbitBinderApplicationTests {
|
||||
|
||||
static Log LOG = LogFactory.getLog(ZipSplitRabbitBinderApplicationTests.class);
|
||||
|
||||
static BlockingQueue<String> DATA_SINK = new LinkedBlockingQueue<>();
|
||||
|
||||
@DynamicPropertySource
|
||||
static void testProperties(DynamicPropertyRegistry registry) {
|
||||
registry.add("file.supplier.directory", () -> new ClassPathResource("/dirWithZips").getPath());
|
||||
}
|
||||
@Container
|
||||
@ServiceConnection
|
||||
static RabbitMQContainer rabbitContainer = new RabbitMQContainer(DockerImageName.parse("rabbitmq:latest"));
|
||||
|
||||
@Test
|
||||
void zippedFilesAreSplittedToRabbitBinding() throws InterruptedException {
|
||||
@@ -51,22 +47,11 @@ class ZipSplitRabbitBinderApplicationTests {
|
||||
}
|
||||
}
|
||||
|
||||
@org.springframework.boot.test.context.TestConfiguration(proxyBeanMethods = false)
|
||||
static class TestConfiguration {
|
||||
|
||||
@Bean
|
||||
@ServiceConnection
|
||||
RabbitMQContainer rabbitContainer() {
|
||||
return new RabbitMQContainer(DockerImageName.parse("rabbitmq:latest"));
|
||||
}
|
||||
|
||||
@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);
|
||||
}
|
||||
|
||||
@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);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user