diff --git a/settings.gradle b/settings.gradle index 49a22913..0353aa87 100644 --- a/settings.gradle +++ b/settings.gradle @@ -29,4 +29,5 @@ include 'spring-pulsar-spring-boot-starter' include 'spring-pulsar-reactive-spring-boot-starter' include 'spring-pulsar-sample-apps:sample-app1' include 'spring-pulsar-sample-apps:sample-app2' +include 'spring-pulsar-sample-apps:sample-reactive' include 'spring-pulsar-docs' diff --git a/spring-pulsar-docs/build.gradle b/spring-pulsar-docs/build.gradle index dae0db96..a66d7d3d 100644 --- a/spring-pulsar-docs/build.gradle +++ b/spring-pulsar-docs/build.gradle @@ -28,7 +28,9 @@ task aggregatedJavadoc(type: Javadoc) { title = "${rootProject.description} ${version} API" project.rootProject.gradle.projectsEvaluated { - Set excludedProjects = ['spring-pulsar-sample-apps:sample-app1', 'spring-pulsar-sample-apps:sample-app2'] + Set excludedProjects = ['spring-pulsar-sample-apps:sample-app1', + 'spring-pulsar-sample-apps:sample-app2', + 'spring-pulsar-sample-apps:sample-reactive'] Set publishedProjects = rootProject.subprojects.findAll { it != project} .findAll { it.plugins.hasPlugin(JavaPlugin) && it.plugins.hasPlugin(MavenPublishPlugin) } .findAll { !excludedProjects.contains(it.name) } diff --git a/spring-pulsar-sample-apps/sample-reactive/build.gradle b/spring-pulsar-sample-apps/sample-reactive/build.gradle new file mode 100644 index 00000000..500559a0 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-reactive/build.gradle @@ -0,0 +1,23 @@ +plugins { + id 'org.springframework.pulsar.spring-module' + id 'org.springframework.boot' version '3.0.0-SNAPSHOT' +} + +description = 'Reactive Spring Pulsar Sample Application' + +dependencies { + api project(':spring-pulsar-reactive-spring-boot-starter') + implementation 'com.google.code.findbugs:jsr305' +} + +bootRun { + jvmArgs = [ + "--add-opens", "java.base/java.lang=ALL-UNNAMED", + "--add-opens", "java.base/java.util=ALL-UNNAMED", + "--add-opens", "java.base/sun.net=ALL-UNNAMED" + ] +} + +project.afterEvaluate { + project.tasks.artifactoryPublish.enabled(false) +} 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 new file mode 100644 index 00000000..c13510c4 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-reactive/src/main/java/org.springframework.pulsar.example/ReactiveSpringPulsarBootApp.java @@ -0,0 +1,122 @@ +/* + * Copyright 2022 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 java.time.Duration; +import java.util.Collections; + +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.reactive.client.api.MessageResult; +import org.apache.pulsar.reactive.client.api.MutableReactiveMessageConsumerSpec; +import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumer; +import org.apache.pulsar.reactive.client.api.ReactivePulsarClient; +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.core.reactive.DefaultReactivePulsarConsumerFactory; +import org.springframework.pulsar.core.reactive.ReactivePulsarConsumerFactory; +import org.springframework.pulsar.core.reactive.ReactivePulsarSenderTemplate; + +import reactor.core.publisher.Flux; + +@SpringBootApplication +public class ReactiveSpringPulsarBootApp { + + public static void main(String[] args) { + SpringApplication.run(ReactiveSpringPulsarBootApp.class, args); + } + + /** + * Sends POJO messages with a reactive template and receives them with a reactive + * consumer. + */ + @Configuration(proxyBeanMethods = false) + static class ReactiveSenderWithReactiveConsumer implements ApplicationListener { + + private final Logger logger = LoggerFactory.getLogger(this.getClass()); + + @Autowired + private ReactivePulsarSenderTemplate reactivePulsarTemplate; + + // TODO remove this once the auto-config is available + @Bean + ReactivePulsarConsumerFactory reactivePulsarConsumerFactory(ReactivePulsarClient reactivePulsarClient) { + MutableReactiveMessageConsumerSpec spec = new MutableReactiveMessageConsumerSpec(); + spec.setTopicNames(Collections.singletonList("sample-reactive-topic1")); + spec.setSubscriptionName("sample-reactive-sub1"); + spec.setConsumerName("sample-reactive-consumer1"); + return new DefaultReactivePulsarConsumerFactory<>(reactivePulsarClient, spec); + } + + @Bean + ApplicationRunner listenPojo(ReactivePulsarConsumerFactory reactiveConsumerFactory) { + return args -> { + ReactiveMessageConsumer messageConsumer = reactiveConsumerFactory + .createConsumer(Schema.JSON(Foo.class)); + messageConsumer + .consumeMany((messageFlux) -> messageFlux.map( + (message) -> MessageResult.acknowledge(message.getMessageId(), message.getValue()))) + .take(Duration.ofSeconds(10)).subscribe((msg) -> this.logger.info("Received: {}", msg)); + }; + } + + @Override + public void onApplicationEvent(ApplicationReadyEvent event) { + this.reactivePulsarTemplate.setSchema(Schema.JSON(Foo.class)); + this.reactivePulsarTemplate + .send("sample-reactive-topic1", Flux.range(0, 10).map((i) -> new Foo("Foo-" + i, "Bar-" + i))) + .subscribe(); + } + + } + + /** + * Sends simple messages with a reactive template and receives them with an imperative + * listener. + */ + @Configuration(proxyBeanMethods = false) + static class ReactiveSenderWithImperativeListener { + + private final Logger logger = LoggerFactory.getLogger(this.getClass()); + + @Bean + ApplicationRunner sendSimple(ReactivePulsarSenderTemplate reactivePulsarTemplate) { + return args -> reactivePulsarTemplate + .send("sample-reactive-topic2", Flux.range(0, 10).map((i) -> "msg-from-sendSimple-" + i)) + .subscribe(); + } + + @PulsarListener(subscriptionName = "sample-reactive-sub2", topics = "sample-reactive-topic2") + void listenSimple(String message) { + this.logger.info("Received: {}", message); + } + + } + + 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/org.springframework.pulsar.example/package-info.java new file mode 100644 index 00000000..a30ad860 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-reactive/src/main/java/org.springframework.pulsar.example/package-info.java @@ -0,0 +1,9 @@ +/** + * Package containing sample apps for the framework. + */ +@NonNullApi +@NonNullFields +package org.springframework.pulsar.example; + +import org.springframework.lang.NonNullApi; +import org.springframework.lang.NonNullFields; 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 new file mode 100644 index 00000000..8ede94de --- /dev/null +++ b/spring-pulsar-sample-apps/sample-reactive/src/main/resources/application.yml @@ -0,0 +1,3 @@ +logging: + level: + org.apache.pulsar: warn diff --git a/src/checkstyle/checkstyle-suppressions.xml b/src/checkstyle/checkstyle-suppressions.xml index 912377c7..ff8ca884 100644 --- a/src/checkstyle/checkstyle-suppressions.xml +++ b/src/checkstyle/checkstyle-suppressions.xml @@ -7,6 +7,7 @@ +