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

[Samples] Make sample-pulsar-reader testable

Browse Source
This commit is contained in:
Chris Bono
2024-01-14 15:32:42 -06:00
committed by Chris Bono
parent 186e2397d2
commit 12334ffcfc
8 changed files with 200 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
spring-pulsar-sample-apps/sample-pulsar-reader/build.gradle
View File

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

7
spring-pulsar-sample-apps/sample-pulsar-reader/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'

88
spring-pulsar-sample-apps/sample-pulsar-reader/src/main/java/com/example/SpringPulsarReaderBootApp.java Normal file
View File

@@ -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<String> 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<Foo> 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) {
}
}

2
spring-pulsar-sample-apps/sample-pulsar-reader/src/main/java/reader/app/package-info.java → spring-pulsar-sample-apps/sample-pulsar-reader/src/main/java/com/example/package-info.java
View File

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

58
spring-pulsar-sample-apps/sample-pulsar-reader/src/main/java/reader/app/SpringPulsarReaderBootApp.java
View File

@@ -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<String> and a PulsarReader.
*/
@Bean
ApplicationRunner runner1(PulsarTemplate<String> 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);
}
}

6
spring-pulsar-sample-apps/sample-pulsar-reader/src/main/resources/application.yml
View File

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

71
spring-pulsar-sample-apps/sample-pulsar-reader/src/test/java/com/example/SpringPulsarReaderBootAppTests.java Normal file
View File

@@ -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<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("++++++READ %s------".formatted(msg));
});
Awaitility.waitAtMost(Duration.ofSeconds(15))
.untilAsserted(() -> assertThat(output).contains(expectedOutput));
}
}

14
spring-pulsar-sample-apps/sample-pulsar-reader/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>