Spring/spring-pulsar
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[Samples] Make sample-reactive testable

Browse Source
This commit is contained in:
Chris Bono
2024-01-14 21:48:09 -06:00
committed by Chris Bono
parent 3f9fad42e8
commit bbdba009b4
8 changed files with 242 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
spring-pulsar-sample-apps/sample-reactive/build.gradle
View File

@@ -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'
}

7
spring-pulsar-sample-apps/sample-reactive/compose.yaml Normal file
View File

@@ -0,0 +1,7 @@
services:
pulsar:
image: 'apachepulsar/pulsar:3.1.2'
ports:
- '6650'
- '8080'
command: 'bin/pulsar standalone'

131
spring-pulsar-sample-apps/sample-reactive/src/main/java/com/example/ReactiveSpringPulsarBootApp.java Normal file
View File

@@ -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<String> 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<Void> 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<Foo> 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<MessageResult<Void>> listenStreaming(Flux<Message<Foo>> messages) {
return messages
.doOnNext((msg) -> LOG.info("++++++CONSUME {}------", msg.getValue()))
.map(MessageResult::acknowledge);
}
}
@Configuration(proxyBeanMethods = false)
static class ConsumerCustomizerConfig {
@Bean
ReactivePulsarListenerMessageConsumerBuilderCustomizer<String> 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<String> 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) {
}
}

2
spring-pulsar-sample-apps/sample-reactive/src/main/java/org.springframework.pulsar.example/package-info.java → spring-pulsar-sample-apps/sample-reactive/src/main/java/com/example/package-info.java
View File

@@ -3,7 +3,7 @@
*/
@NonNullApi
@NonNullFields
package org.springframework.pulsar.example;
package com.example;
import org.springframework.lang.NonNullApi;
import org.springframework.lang.NonNullFields;

143
spring-pulsar-sample-apps/sample-reactive/src/main/java/org.springframework.pulsar.example/ReactiveSpringPulsarBootApp.java
View File

@@ -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<ApplicationReadyEvent> {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
@Autowired
private ReactivePulsarTemplate<String> 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<Void> 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<ApplicationReadyEvent> {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
@Autowired
private ReactivePulsarTemplate<Foo> reactivePulsarTemplate;
@Override
public void onApplicationEvent(ApplicationReadyEvent event) {
Schema<Foo> 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<MessageResult<Void>> listenStreaming(Flux<Message<Foo>> 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<String> 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<String> 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) {
}
}

8
spring-pulsar-sample-apps/sample-reactive/src/main/resources/application.yml
View File

@@ -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

75
spring-pulsar-sample-apps/sample-reactive/src/test/java/com/example/ReactiveSpringPulsarBootAppTests.java Normal file
View File

@@ -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<Integer, Object> 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));
}
}

14
spring-pulsar-sample-apps/sample-reactive/src/test/resources/logback-test.xml Normal file
View File

@@ -0,0 +1,14 @@
<configuration>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n</pattern>
</encoder>
</appender>
<root level="WARN">
<appender-ref ref="STDOUT"/>
</root>
<logger name="com.example" level="INFO"/>
<logger name="com.github.dockerjava" level="ERROR"/>
<logger name="org.apache.pulsar.common.util.netty" level="ERROR" />
<logger name="org.testcontainers" level="ERROR"/>
</configuration>