diff --git a/spring-pulsar-sample-apps/sample-pulsar-reader/build.gradle b/spring-pulsar-sample-apps/sample-pulsar-reader/build.gradle index a3ab9a65..63679dd8 100644 --- a/spring-pulsar-sample-apps/sample-pulsar-reader/build.gradle +++ b/spring-pulsar-sample-apps/sample-pulsar-reader/build.gradle @@ -18,8 +18,19 @@ def pulsarVersion = versionCatalog.findVersion("pulsar").orElseThrow().displayNa ext['spring-pulsar.version'] = "${project.version}" ext['pulsar.version'] = "${pulsarVersion}" + dependencies { - implementation "org.springframework.boot:spring-boot-starter-pulsar" + implementation 'org.springframework.boot:spring-boot-starter-pulsar' + implementation 'org.springframework.boot:spring-boot-starter-actuator' + developmentOnly 'org.springframework.boot:spring-boot-docker-compose' + testImplementation project(':spring-pulsar-test') + testRuntimeOnly 'ch.qos.logback:logback-classic' + testRuntimeOnly 'org.junit.platform:junit-platform-launcher' + testImplementation 'org.awaitility:awaitility' + 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 { @@ -34,4 +45,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-pulsar-reader/compose.yaml b/spring-pulsar-sample-apps/sample-pulsar-reader/compose.yaml new file mode 100644 index 00000000..39df55a5 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-pulsar-reader/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-pulsar-reader/src/main/java/com/example/SpringPulsarReaderBootApp.java b/spring-pulsar-sample-apps/sample-pulsar-reader/src/main/java/com/example/SpringPulsarReaderBootApp.java new file mode 100644 index 00000000..49e7170d --- /dev/null +++ b/spring-pulsar-sample-apps/sample-pulsar-reader/src/main/java/com/example/SpringPulsarReaderBootApp.java @@ -0,0 +1,88 @@ +/* + * Copyright 2023-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.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.PulsarReader; +import org.springframework.pulsar.core.PulsarTemplate; + +@SpringBootApplication +public class SpringPulsarReaderBootApp { + + private static final Logger LOG = LoggerFactory.getLogger(SpringPulsarReaderBootApp.class); + + public static void main(String[] args) { + SpringApplication.run(SpringPulsarReaderBootApp.class, args); + } + + @Configuration(proxyBeanMethods = false) + static class ProduceAndReadWithPrimitiveMessageType { + + private static final String TOPIC = "produce-read-primitive"; + + @Bean + ApplicationRunner sendPrimitiveMessagesToPulsarTopic(PulsarTemplate template) { + return (args) -> { + for (int i = 0; i < 10; i++) { + var msg = "ProduceAndReadWithPrimitiveMessageType:" + i; + template.send(TOPIC, msg); + LOG.info("++++++PRODUCE {}------", msg); + } + }; + } + + @PulsarReader(topics = TOPIC, startMessageId = "earliest") + void readPrimitiveMessagesFromPulsarTopic(String msg) { + LOG.info("++++++READ {}------", msg); + } + + } + + @Configuration(proxyBeanMethods = false) + static class ProduceAndReadWithComplexMessageType { + + private static final String TOPIC = "produce-read-complex"; + + @Bean + ApplicationRunner sendComplexMessagesToPulsarTopic(PulsarTemplate template) { + return (args) -> { + for (int i = 0; i < 10; i++) { + var msg = new Foo("ProduceAndReadWithComplexMessageType", i); + template.send(TOPIC, msg); + LOG.info("++++++PRODUCE {}------", msg); + } + }; + } + + @PulsarReader(topics = TOPIC, startMessageId = "earliest") + void readComplexMessagesFromPulsarTopic(Foo msg) { + LOG.info("++++++READ {}------", msg); + } + + } + + record Foo(String name, Integer value) { + } + +} diff --git a/spring-pulsar-sample-apps/sample-pulsar-reader/src/main/java/reader/app/package-info.java b/spring-pulsar-sample-apps/sample-pulsar-reader/src/main/java/com/example/package-info.java similarity index 89% rename from spring-pulsar-sample-apps/sample-pulsar-reader/src/main/java/reader/app/package-info.java rename to spring-pulsar-sample-apps/sample-pulsar-reader/src/main/java/com/example/package-info.java index c03ac2f1..5f2a90d1 100644 --- a/spring-pulsar-sample-apps/sample-pulsar-reader/src/main/java/reader/app/package-info.java +++ b/spring-pulsar-sample-apps/sample-pulsar-reader/src/main/java/com/example/package-info.java @@ -3,7 +3,7 @@ */ @NonNullApi @NonNullFields -package reader.app; +package com.example; import org.springframework.lang.NonNullApi; import org.springframework.lang.NonNullFields; diff --git a/spring-pulsar-sample-apps/sample-pulsar-reader/src/main/java/reader/app/SpringPulsarReaderBootApp.java b/spring-pulsar-sample-apps/sample-pulsar-reader/src/main/java/reader/app/SpringPulsarReaderBootApp.java deleted file mode 100644 index 478b1cf0..00000000 --- a/spring-pulsar-sample-apps/sample-pulsar-reader/src/main/java/reader/app/SpringPulsarReaderBootApp.java +++ /dev/null @@ -1,58 +0,0 @@ -/* - * Copyright 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 reader.app; - -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.pulsar.annotation.PulsarReader; -import org.springframework.pulsar.core.PulsarTemplate; - -@SpringBootApplication -public class SpringPulsarReaderBootApp { - - private static final Logger logger = LoggerFactory.getLogger(SpringPulsarReaderBootApp.class); - - public static void main(String[] args) { - SpringApplication.run(SpringPulsarReaderBootApp.class, args); - } - - /* - * Basic publisher using PulsarTemplate and a PulsarReader. - */ - @Bean - ApplicationRunner runner1(PulsarTemplate pulsarTemplate) { - - String topic1 = "pulsar-reader-demo-topic"; - - return args -> { - for (int i = 0; i < 10; i++) { - pulsarTemplate.send(topic1, "This is message " + (i + 1)); - } - }; - } - - @PulsarReader(id = "my-id", topics = "pulsar-reader-demo-topic", startMessageId = "earliest") - void read(String message) { - logger.info(message); - } - -} diff --git a/spring-pulsar-sample-apps/sample-pulsar-reader/src/main/resources/application.yml b/spring-pulsar-sample-apps/sample-pulsar-reader/src/main/resources/application.yml index 8b137891..94bca695 100644 --- a/spring-pulsar-sample-apps/sample-pulsar-reader/src/main/resources/application.yml +++ b/spring-pulsar-sample-apps/sample-pulsar-reader/src/main/resources/application.yml @@ -1 +1,5 @@ - +spring: + docker: + compose: + # when run from Intellij via "Run" button, path must be set from project root + file: spring-pulsar-sample-apps/sample-pulsar-reader/compose.yaml diff --git a/spring-pulsar-sample-apps/sample-pulsar-reader/src/test/java/com/example/SpringPulsarReaderBootAppTests.java b/spring-pulsar-sample-apps/sample-pulsar-reader/src/test/java/com/example/SpringPulsarReaderBootAppTests.java new file mode 100644 index 00000000..94c09368 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-pulsar-reader/src/test/java/com/example/SpringPulsarReaderBootAppTests.java @@ -0,0 +1,71 @@ +/* + * 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.SpringPulsarReaderBootApp.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 SpringPulsarReaderBootAppTests 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 produceConsumeWithPrimitiveMessageType(CapturedOutput output) { + verifyProduceConsume(output,10, (i) -> "ProduceAndReadWithPrimitiveMessageType:" + i); + } + + @Test + void produceConsumeWithComplexMessageType(CapturedOutput output) { + verifyProduceConsume(output,10, + (i) -> new Foo("ProduceAndReadWithComplexMessageType", 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("++++++READ %s------".formatted(msg)); + }); + Awaitility.waitAtMost(Duration.ofSeconds(15)) + .untilAsserted(() -> assertThat(output).contains(expectedOutput)); + } +} diff --git a/spring-pulsar-sample-apps/sample-pulsar-reader/src/test/resources/logback-test.xml b/spring-pulsar-sample-apps/sample-pulsar-reader/src/test/resources/logback-test.xml new file mode 100644 index 00000000..97f7e370 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-pulsar-reader/src/test/resources/logback-test.xml @@ -0,0 +1,14 @@ + + + + %d{HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n + + + + + + + + + +