From bbdba009b481fc04f0cd04848adf6c4e0194b7dd Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Sun, 14 Jan 2024 21:48:09 -0600 Subject: [PATCH] [Samples] Make sample-reactive testable --- .../sample-reactive/build.gradle | 9 ++ .../sample-reactive/compose.yaml | 7 + .../example/ReactiveSpringPulsarBootApp.java | 131 ++++++++++++++++ .../example}/package-info.java | 2 +- .../ReactiveSpringPulsarBootApp.java | 143 ------------------ .../src/main/resources/application.yml | 8 +- .../ReactiveSpringPulsarBootAppTests.java | 75 +++++++++ .../src/test/resources/logback-test.xml | 14 ++ 8 files changed, 242 insertions(+), 147 deletions(-) create mode 100644 spring-pulsar-sample-apps/sample-reactive/compose.yaml create mode 100644 spring-pulsar-sample-apps/sample-reactive/src/main/java/com/example/ReactiveSpringPulsarBootApp.java rename spring-pulsar-sample-apps/sample-reactive/src/main/java/{org.springframework.pulsar.example => com/example}/package-info.java (80%) delete mode 100644 spring-pulsar-sample-apps/sample-reactive/src/main/java/org.springframework.pulsar.example/ReactiveSpringPulsarBootApp.java create mode 100644 spring-pulsar-sample-apps/sample-reactive/src/test/java/com/example/ReactiveSpringPulsarBootAppTests.java create mode 100644 spring-pulsar-sample-apps/sample-reactive/src/test/resources/logback-test.xml diff --git a/spring-pulsar-sample-apps/sample-reactive/build.gradle b/spring-pulsar-sample-apps/sample-reactive/build.gradle index 27ab530a..135d681e 100644 --- a/spring-pulsar-sample-apps/sample-reactive/build.gradle +++ b/spring-pulsar-sample-apps/sample-reactive/build.gradle @@ -22,6 +22,13 @@ ext['pulsar-reactive.version'] = "${pulsarReactiveVersion}" dependencies { implementation "org.springframework.boot:spring-boot-starter-pulsar-reactive" + developmentOnly 'org.springframework.boot:spring-boot-docker-compose' + testImplementation project(':spring-pulsar-test') + testRuntimeOnly 'ch.qos.logback:logback-classic' + testImplementation "org.springframework.boot:spring-boot-starter-test" + testImplementation "org.springframework.boot:spring-boot-testcontainers" + testImplementation 'org.testcontainers:junit-jupiter' + testImplementation 'org.testcontainers:pulsar' } test { @@ -36,4 +43,6 @@ bootRun { "--add-opens", "java.base/java.util=ALL-UNNAMED", "--add-opens", "java.base/sun.net=ALL-UNNAMED" ] + // when run from command line, path must be set relative to module dir + systemProperty 'spring.docker.compose.file', 'compose.yaml' } diff --git a/spring-pulsar-sample-apps/sample-reactive/compose.yaml b/spring-pulsar-sample-apps/sample-reactive/compose.yaml new file mode 100644 index 00000000..39df55a5 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-reactive/compose.yaml @@ -0,0 +1,7 @@ +services: + pulsar: + image: 'apachepulsar/pulsar:3.1.2' + ports: + - '6650' + - '8080' + command: 'bin/pulsar standalone' diff --git a/spring-pulsar-sample-apps/sample-reactive/src/main/java/com/example/ReactiveSpringPulsarBootApp.java b/spring-pulsar-sample-apps/sample-reactive/src/main/java/com/example/ReactiveSpringPulsarBootApp.java new file mode 100644 index 00000000..5f8f1abe --- /dev/null +++ b/spring-pulsar-sample-apps/sample-reactive/src/main/java/com/example/ReactiveSpringPulsarBootApp.java @@ -0,0 +1,131 @@ +/* + * Copyright 2022-2024 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.example; + +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.SubscriptionInitialPosition; +import org.apache.pulsar.common.schema.SchemaType; +import org.apache.pulsar.reactive.client.api.MessageResult; +import org.apache.pulsar.reactive.client.api.MessageSpec; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.springframework.boot.ApplicationRunner; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.pulsar.annotation.PulsarListener; +import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListener; +import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListenerMessageConsumerBuilderCustomizer; +import org.springframework.pulsar.reactive.core.ReactivePulsarTemplate; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +@SpringBootApplication +public class ReactiveSpringPulsarBootApp { + + private static final Logger LOG = LoggerFactory.getLogger(ReactiveSpringPulsarBootApp.class); + + public static void main(String[] args) { + SpringApplication.run(ReactiveSpringPulsarBootApp.class, args); + } + + @Configuration(proxyBeanMethods = false) + static class ReactiveTemplateWithSimpleReactiveListener { + + private static final String TOPIC = "sample-reactive-topic1"; + + @Bean + ApplicationRunner sendPrimitiveMessagesToPulsarTopic(ReactivePulsarTemplate template) { + return (args) -> Flux.range(0, 10) + .map((i) -> MessageSpec.of("ReactiveTemplateWithSimpleReactiveListener:" + i)) + .as(messages -> template.send(TOPIC, messages)) + .doOnNext((msr) -> LOG.info("++++++PRODUCE {}------", msr.getMessageSpec().getValue())) + .subscribe(); + } + + @ReactivePulsarListener(topics = TOPIC, consumerCustomizer = "subscriptionInitialPositionEarliest") + public Mono listenSimple(String msg) { + LOG.info("++++++CONSUME {}------", msg); + return Mono.empty(); + } + + } + + @Configuration(proxyBeanMethods = false) + static class ReactiveTemplateWithStreamingReactiveListener { + + private static final String TOPIC = "sample-reactive-topic2"; + + @Bean + ApplicationRunner sendComplexMessagesToPulsarTopic(ReactivePulsarTemplate template) { + var schema = Schema.JSON(Foo.class); + return (args) -> Flux.range(0, 10) + .map((i) -> MessageSpec.of(new Foo("Foo-" + i, "Bar-" + i))) + .as(messages -> template.send(TOPIC, messages, schema)) + .doOnNext((msr) -> LOG.info("++++++PRODUCE {}------", msr.getMessageSpec().getValue())) + .subscribe(); + } + + @ReactivePulsarListener(topics = TOPIC, stream = true, schemaType = SchemaType.JSON, + consumerCustomizer = "subscriptionInitialPositionEarliest") + public Flux> listenStreaming(Flux> messages) { + return messages + .doOnNext((msg) -> LOG.info("++++++CONSUME {}------", msg.getValue())) + .map(MessageResult::acknowledge); + } + + } + + @Configuration(proxyBeanMethods = false) + static class ConsumerCustomizerConfig { + + @Bean + ReactivePulsarListenerMessageConsumerBuilderCustomizer subscriptionInitialPositionEarliest() { + return b -> b.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest); + } + + } + + @Configuration(proxyBeanMethods = false) + static class ReactiveTemplateWithImperativeListener { + + private static final String TOPIC = "sample-reactive-topic3"; + + @Bean + ApplicationRunner sendMessagesToPulsarTopic(ReactivePulsarTemplate template) { + return (args) -> Flux.range(0, 10) + .map((i) -> MessageSpec.of("ReactiveTemplateWithImperativeListener:" + i)) + .as(messages -> template.send(TOPIC, messages)) + .doOnNext((msr) -> LOG.info("++++++PRODUCE {}------", msr.getMessageSpec().getValue())) + .subscribe(); + } + + @PulsarListener(topics = TOPIC) + void listenSimple(String msg) { + LOG.info("++++++CONSUME {}------", msg); + } + + } + + record Foo(String foo, String bar) { + } + +} diff --git a/spring-pulsar-sample-apps/sample-reactive/src/main/java/org.springframework.pulsar.example/package-info.java b/spring-pulsar-sample-apps/sample-reactive/src/main/java/com/example/package-info.java similarity index 80% rename from spring-pulsar-sample-apps/sample-reactive/src/main/java/org.springframework.pulsar.example/package-info.java rename to spring-pulsar-sample-apps/sample-reactive/src/main/java/com/example/package-info.java index a30ad860..12a7ef97 100644 --- a/spring-pulsar-sample-apps/sample-reactive/src/main/java/org.springframework.pulsar.example/package-info.java +++ b/spring-pulsar-sample-apps/sample-reactive/src/main/java/com/example/package-info.java @@ -3,7 +3,7 @@ */ @NonNullApi @NonNullFields -package org.springframework.pulsar.example; +package com.example; import org.springframework.lang.NonNullApi; import org.springframework.lang.NonNullFields; diff --git a/spring-pulsar-sample-apps/sample-reactive/src/main/java/org.springframework.pulsar.example/ReactiveSpringPulsarBootApp.java b/spring-pulsar-sample-apps/sample-reactive/src/main/java/org.springframework.pulsar.example/ReactiveSpringPulsarBootApp.java deleted file mode 100644 index 961c6abc..00000000 --- a/spring-pulsar-sample-apps/sample-reactive/src/main/java/org.springframework.pulsar.example/ReactiveSpringPulsarBootApp.java +++ /dev/null @@ -1,143 +0,0 @@ -/* - * Copyright 2022-2023 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. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.pulsar.example; - -import org.apache.pulsar.client.api.Message; -import org.apache.pulsar.client.api.Schema; -import org.apache.pulsar.client.api.SubscriptionInitialPosition; -import org.apache.pulsar.common.schema.SchemaType; -import org.apache.pulsar.reactive.client.api.MessageResult; -import org.apache.pulsar.reactive.client.api.MessageSpec; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.ApplicationRunner; -import org.springframework.boot.SpringApplication; -import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.boot.context.event.ApplicationReadyEvent; -import org.springframework.context.ApplicationListener; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.pulsar.annotation.PulsarListener; -import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListener; -import org.springframework.pulsar.reactive.core.ReactiveMessageConsumerBuilderCustomizer; -import org.springframework.pulsar.reactive.core.ReactivePulsarTemplate; - -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; - -@SpringBootApplication -public class ReactiveSpringPulsarBootApp { - - public static void main(String[] args) { - SpringApplication.run(ReactiveSpringPulsarBootApp.class, args); - } - - /** - * Sends string messages with a reactive template and receives them with a simple - * reactive listener. - */ - @Configuration(proxyBeanMethods = false) - static class ReactiveTemplateWithSimpleReactiveListener implements ApplicationListener { - - private final Logger logger = LoggerFactory.getLogger(this.getClass()); - - @Autowired - private ReactivePulsarTemplate reactivePulsarTemplate; - - @Override - public void onApplicationEvent(ApplicationReadyEvent event) { - Flux.range(0, 10).map((i) -> MessageSpec.of("sample-message-" + i)) - .as(messages -> this.reactivePulsarTemplate.send("sample-reactive-topic1", messages)).subscribe(); - } - - @ReactivePulsarListener(subscriptionName = "sample-reactive-sub1", topics = "sample-reactive-topic1", - consumerCustomizer = "subscriptionInitialPositionEarliest") - public Mono listenSimple(String msg) { - this.logger.info("Simple reactive listener received: {}", msg); - return Mono.empty(); - } - - } - - /** - * Sends POJO messages with a reactive template and receives them with a reactive - * streaming listener. - */ - @Configuration(proxyBeanMethods = false) - static class ReactiveTemplateWithStreamingReactiveListener implements ApplicationListener { - - private final Logger logger = LoggerFactory.getLogger(this.getClass()); - - @Autowired - private ReactivePulsarTemplate reactivePulsarTemplate; - - @Override - public void onApplicationEvent(ApplicationReadyEvent event) { - Schema schema = Schema.JSON(Foo.class); - Flux.range(0, 10).map((i) -> MessageSpec.of(new Foo("Foo-" + i, "Bar-" + i))) - .as(messages -> this.reactivePulsarTemplate.send("sample-reactive-topic2", messages, schema)) - .subscribe(); - } - - @ReactivePulsarListener(subscriptionName = "sample-reactive-sub2", topics = "sample-reactive-topic2", - stream = true, schemaType = SchemaType.JSON, consumerCustomizer = "subscriptionInitialPositionEarliest") - public Flux> listenStreaming(Flux> messages) { - return messages - .doOnNext((msg) -> this.logger.info("Streaming reactive listener received: {}", msg.getValue())) - .map(MessageResult::acknowledge); - } - - } - - @Configuration(proxyBeanMethods = false) - static class ConsumerCustomizerConfig { - - @Bean - ReactiveMessageConsumerBuilderCustomizer subscriptionInitialPositionEarliest() { - return b -> b.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest); - } - - } - - /** - * Sends simple messages with a reactive template and receives them with an imperative - * listener. - */ - @Configuration(proxyBeanMethods = false) - static class ReactiveTemplateWithImperativeListener { - - private final Logger logger = LoggerFactory.getLogger(this.getClass()); - - @Bean - ApplicationRunner sendSimple(ReactivePulsarTemplate reactivePulsarTemplate) { - return args -> Flux.range(0, 10).map((i) -> MessageSpec.of("msg-from-sendSimple-" + i)) - .as(messages -> reactivePulsarTemplate.send("sample-reactive-topic3", messages)).subscribe(); - } - - @PulsarListener(subscriptionName = "sample-reactive-sub3", topics = "sample-reactive-topic3") - void listenSimple(String message) { - this.logger.info("Imperative listener received: {}", message); - } - - } - - record Foo(String foo, String bar) { - } - -} diff --git a/spring-pulsar-sample-apps/sample-reactive/src/main/resources/application.yml b/spring-pulsar-sample-apps/sample-reactive/src/main/resources/application.yml index 8ede94de..5cfecb5a 100644 --- a/spring-pulsar-sample-apps/sample-reactive/src/main/resources/application.yml +++ b/spring-pulsar-sample-apps/sample-reactive/src/main/resources/application.yml @@ -1,3 +1,5 @@ -logging: - level: - org.apache.pulsar: warn +spring: + docker: + compose: + # when run from Intellij via "Run" button, path must be set from project root + file: spring-pulsar-sample-apps/sample-reactive/compose.yaml diff --git a/spring-pulsar-sample-apps/sample-reactive/src/test/java/com/example/ReactiveSpringPulsarBootAppTests.java b/spring-pulsar-sample-apps/sample-reactive/src/test/java/com/example/ReactiveSpringPulsarBootAppTests.java new file mode 100644 index 00000000..d82bc943 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-reactive/src/test/java/com/example/ReactiveSpringPulsarBootAppTests.java @@ -0,0 +1,75 @@ +/* + * Copyright 2012-2024 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.example; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.function.Function; +import java.util.stream.IntStream; + +import com.example.ReactiveSpringPulsarBootApp.Foo; +import org.awaitility.Awaitility; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; + +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.system.CapturedOutput; +import org.springframework.boot.test.system.OutputCaptureExtension; +import org.springframework.pulsar.test.support.PulsarTestContainerSupport; +import org.springframework.test.context.DynamicPropertyRegistry; +import org.springframework.test.context.DynamicPropertySource; + +import static org.assertj.core.api.Assertions.assertThat; + +@SpringBootTest +@ExtendWith(OutputCaptureExtension.class) +class ReactiveSpringPulsarBootAppTests implements PulsarTestContainerSupport { + + @DynamicPropertySource + static void pulsarProperties(DynamicPropertyRegistry registry) { + registry.add("spring.pulsar.client.service-url", PULSAR_CONTAINER::getPulsarBrokerUrl); + registry.add("spring.pulsar.admin.service-url", PULSAR_CONTAINER::getHttpServiceUrl); + } + + @Test + void reactiveTemplateWithSimpleReactiveListener(CapturedOutput output) { + verifyProduceConsume(output,10, (i) -> "ReactiveTemplateWithSimpleReactiveListener:" + i); + } + + @Test + void reactiveTemplateWithStreamingReactiveListener(CapturedOutput output) { + verifyProduceConsume(output,10, (i) -> new Foo("Foo-" + i, "Bar-" + i)); + } + + @Test + void reactiveTemplateWithImperativeListener(CapturedOutput output) { + verifyProduceConsume(output,10, (i) -> "ReactiveTemplateWithImperativeListener:" + i); + } + + private void verifyProduceConsume(CapturedOutput output, int numExpectedMessages, + Function expectedMessageFactory) { + List < String > expectedOutput = new ArrayList<>(); + IntStream.range(0, numExpectedMessages).forEachOrdered((i) -> { + var msg = expectedMessageFactory.apply(i); + expectedOutput.add("++++++PRODUCE %s------".formatted(msg)); + expectedOutput.add("++++++CONSUME %s------".formatted(msg)); + }); + Awaitility.waitAtMost(Duration.ofSeconds(15)) + .untilAsserted(() -> assertThat(output).contains(expectedOutput)); + } +} diff --git a/spring-pulsar-sample-apps/sample-reactive/src/test/resources/logback-test.xml b/spring-pulsar-sample-apps/sample-reactive/src/test/resources/logback-test.xml new file mode 100644 index 00000000..97f7e370 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-reactive/src/test/resources/logback-test.xml @@ -0,0 +1,14 @@ + + + + %d{HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n + + + + + + + + + +