diff --git a/README.adoc b/README.adoc index 372f473e..5875b445 100644 --- a/README.adoc +++ b/README.adoc @@ -14,6 +14,10 @@ Most of the ideas in this project are borrowed from the Spring for Apache Kafka **Apache Pulsar** - 2.10.0 +**Spring Boot** - 3.0.0 + +**Spring Framework** - 6.0.0 + ``` ./gradlew clean build ``` diff --git a/build.gradle b/build.gradle index 112849c7..301a5836 100644 --- a/build.gradle +++ b/build.gradle @@ -1,10 +1,4 @@ buildscript { -// repositories { -// mavenCentral() -// maven { url 'https://plugins.gradle.org/m2' } -// maven { url 'https://repo.spring.io/plugins-release' } -// mavenLocal() -// } repositories { mavenCentral() gradlePluginPortal() @@ -17,7 +11,7 @@ plugins { id 'project-report' id 'idea' id 'org.sonarqube' version '2.8' -// id 'org.ajoberstar.grgit' version '4.0.1' apply false + id 'org.ajoberstar.grgit' version '4.0.1' apply false id 'io.spring.nohttp' version '0.0.5.RELEASE' id 'io.spring.dependency-management' version '1.0.10.RELEASE' apply false id 'com.jfrog.artifactory' version '4.18.2' apply false @@ -28,36 +22,46 @@ plugins { apply plugin: 'io.spring.nohttp' -//def gitPresent = new File('.git').exists() +def gitPresent = new File('.git').exists() -//if(gitPresent) { -// apply plugin: 'org.ajoberstar.grgit' -//} +if (gitPresent) { + apply plugin: 'org.ajoberstar.grgit' +} description = 'Spring for Apache Pulsar' ext { -// if (gitPresent) { -// modifiedFiles = -// files(grgit.status().unstaged.modified).filter{ f -> f.name.endsWith('.java') || f.name.endsWith('.kt') } -// } + linkHomepage = 'https://github.com/spring-projects-experimental/spring-pulsar' + linkIssue = 'https://github.com/spring-projects-experimental/spring-pulsar/issues' + linkScmUrl = 'https://github.com/spring-projects-experimental/spring-pulsar' + linkScmConnection = 'https://github.com/spring-projects-experimental/spring-pulsar.git' + linkScmDevConnection = 'git@github.com:spring-projects-experimental/spring-pulsar.git' + docResourcesVersion = '0.2.5' - assertjVersion = '3.21.0' - awaitilityVersion = '4.1.1' + javadocLinks = [ + 'https://docs.oracle.com/en/java/javase/17/docs/api/', + 'https://docs.spring.io/spring-framework/docs/current/javadoc-api/' + ] as String[] + + if (gitPresent) { + modifiedFiles = + files(grgit.status().unstaged.modified).filter{ f -> f.name.endsWith('.java') } + } + + assertjVersion = '3.22.0' + awaitilityVersion = '4.2.0' googleJsr305Version = '3.0.2' hamcrestVersion = '2.2' - hibernateValidationVersion = '6.2.3.Final' - jacksonBomVersion = '2.13.2.20220328' + hibernateValidationVersion = '7.0.4.Final' + jacksonBomVersion = '2.13.3' jaywayJsonPathVersion = '2.6.0' - junit4Version = '4.13.2' junitJupiterVersion = '5.8.2' pulsarVersion = '2.10.0' log4jVersion = '2.17.2' -// micrometerVersion = '2.0.0-SNAPSHOT' - mockitoVersion = '4.0.0' + mockitoVersion = '4.5.1' reactorVersion = '2020.0.17' springBootVersion = '3.0.0-SNAPSHOT' // docs module - springRetryVersion = '1.3.2' + springRetryVersion = '1.3.3' springVersion = '6.0.0-SNAPSHOT' idPrefix = 'pulsar' @@ -105,11 +109,11 @@ allprojects { subprojects { subproject -> apply plugin: 'java-library' apply plugin: 'java' -// apply from: "${rootProject.projectDir}/publish-maven.gradle" + apply from: "${rootProject.projectDir}/publish-maven.gradle" apply plugin: 'eclipse' apply plugin: 'idea' apply plugin: 'jacoco' -// apply plugin: 'checkstyle' + apply plugin: 'checkstyle' java { withJavadocJar() @@ -162,12 +166,9 @@ subprojects { subproject -> testImplementation "org.hamcrest:hamcrest-core:$hamcrestVersion" optionalApi "org.assertj:assertj-core:$assertjVersion" - - testImplementation("org.testcontainers:pulsar:1.17.2") { exclude module: 'log4j-to-slf4j' } - } // enable all compiler warnings; individual projects may customize further @@ -208,15 +209,68 @@ subprojects { subproject -> } } + publishing { + publications { + mavenJava(MavenPublication) { + suppressAllPomMetadataWarnings() + from components.java + pom.withXml { + def pomDeps = asNode().dependencies.first() + subproject.configurations.providedImplementation.allDependencies.each { dep -> + pomDeps.remove(pomDeps.'*'.find { it.artifactId.text() == dep.name }) + pomDeps.appendNode('dependency').with { + it.appendNode('groupId', dep.group) + it.appendNode('artifactId', dep.name) + it.appendNode('version', dep.version) + it.appendNode('scope', 'provided') + } + } + } + } + } + } + + task updateCopyrights { + onlyIf { gitPresent && !System.getenv('GITHUB_ACTION') } + if (gitPresent) { + inputs.files(modifiedFiles.filter { f -> f.path.contains(subproject.name) }) + } + outputs.dir('build') + + doLast { + def now = Calendar.instance.get(Calendar.YEAR) as String + inputs.files.each { file -> + def line + file.withReader { reader -> + while (line = reader.readLine()) { + def matcher = line =~ /Copyright (20\d\d)-?(20\d\d)?/ + if (matcher.count) { + def beginningYear = matcher[0][1] + if (now != beginningYear && now != matcher[0][2]) { + def years = "$beginningYear-$now" + def sourceCode = file.text + sourceCode = sourceCode.replaceFirst(/20\d\d(-20\d\d)?/, years) + file.write(sourceCode) + println "Copyright updated for file: $file" + } + break + } + } + } + } + } + } + + jar { manifest { attributes( 'Implementation-Version': archiveVersion, -// 'Created-By': "JDK ${System.properties['java.version']} (${System.properties['java.specification.vendor']})", + 'Created-By': "JDK ${System.properties['java.version']} (${System.properties['java.specification.vendor']})", 'Implementation-Title': subproject.name, 'Implementation-Vendor-Id': subproject.group, 'Implementation-Vendor': 'Pivotal Software, Inc.', -// 'Implementation-URL': linkHomepage, + 'Implementation-URL': linkHomepage, 'Automatic-Module-Name': subproject.name.replace('-', '.') // for Jigsaw ) } @@ -227,6 +281,10 @@ subprojects { subproject -> into 'META-INF' expand(copyright: new Date().format('yyyy'), version: project.version) } + from("${rootProject.projectDir}") { + include 'LICENSE.txt' + into 'META-INF' + } } tasks.withType(Javadoc) { @@ -262,13 +320,10 @@ project ('spring-pulsar') { optionalApi "com.jayway.jsonpath:json-path:$jaywayJsonPathVersion" optionalApi 'io.projectreactor:reactor-core' -// optionalApi "io.micrometer:micrometer-core:$micrometerVersion" testImplementation 'io.projectreactor:reactor-test' testImplementation "org.mockito:mockito-junit-jupiter:$mockitoVersion" testImplementation "org.hibernate.validator:hibernate-validator:$hibernateValidationVersion" - - testImplementation project (':spring-pulsar-boot-autoconfigure') } } @@ -296,3 +351,14 @@ project ('spring-pulsar-sample-apps') { } } +sonarqube { + properties { + property 'sonar.links.homepage', linkHomepage + property 'sonar.links.ci', linkCi + property 'sonar.links.issue', linkIssue + property 'sonar.links.scm', linkScmUrl + property 'sonar.links.scm_dev', linkScmDevConnection + } +} + + diff --git a/publish-maven.gradle b/publish-maven.gradle new file mode 100644 index 00000000..28af4270 --- /dev/null +++ b/publish-maven.gradle @@ -0,0 +1,52 @@ +apply plugin: 'maven-publish' +apply plugin: 'com.jfrog.artifactory' + +publishing { + publications { + mavenJava(MavenPublication) { + pom { + afterEvaluate { + name = project.description + description = project.description + } + url = linkScmUrl +// organization { +// name = 'Spring IO' +// url = 'https://spring.io/projects/spring-pulsar' +// } + licenses { + license { + name = 'Apache License, Version 2.0' + url = 'https://www.apache.org/licenses/LICENSE-2.0.txt' + distribution = 'repo' + } + } + scm { + url = linkScmUrl + connection = linkScmConnection + developerConnection = linkScmDevConnection + } +// developers { +// +// } + issueManagement { + system = 'GitHub' + url = linkIssue + } + } + versionMapping { + usage('java-api') { + fromResolutionResult() + } + usage('java-runtime') { + fromResolutionResult() + } + } + } + } +} + +artifactoryPublish { + dependsOn build + publications(publishing.publications.mavenJava) +} diff --git a/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java b/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java index 1dd37ab8..20ed964d 100644 --- a/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java +++ b/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java @@ -28,6 +28,8 @@ import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.listener.PulsarContainerProperties; /** + * Configuration for Pulsar annotation-driven support. + * * @author Soby Chacko */ @Configuration(proxyBeanMethods = false) diff --git a/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java b/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java index 60d22fc8..83a41807 100644 --- a/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java +++ b/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java @@ -24,15 +24,17 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Import; -import org.springframework.pulsar.core.DefaultPulsarConsumerFactory; -import org.springframework.pulsar.core.DefaultPulsarProducerFactory; import org.springframework.pulsar.config.PulsarClientConfiguration; import org.springframework.pulsar.config.PulsarClientFactoryBean; +import org.springframework.pulsar.core.DefaultPulsarConsumerFactory; +import org.springframework.pulsar.core.DefaultPulsarProducerFactory; import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.core.PulsarProducerFactory; import org.springframework.pulsar.core.PulsarTemplate; /** + * {@link org.springframework.boot.autoconfigure.EnableAutoConfiguration Auto-configuration} for Apache Pulsar. + * * @author Soby Chacko */ @AutoConfiguration diff --git a/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java b/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java index 8d0833b8..fac31d83 100644 --- a/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java +++ b/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java @@ -33,12 +33,16 @@ import org.apache.pulsar.client.api.ProducerCryptoFailureAction; import org.apache.pulsar.client.api.RegexSubscriptionMode; import org.apache.pulsar.client.api.SubscriptionInitialPosition; import org.apache.pulsar.client.api.SubscriptionType; -import org.apache.pulsar.common.schema.SchemaType; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.boot.context.properties.PropertyMapper; /** + * Configuration properties for Spring for Apache Pulsar. + *

+ * Users should refer to Pulsar documentation for complete descriptions of these + * properties. + * * @author Soby Chacko */ @ConfigurationProperties(prefix = "spring.pulsar") @@ -380,7 +384,7 @@ public class PulsarProperties { private ProducerAccessMode producerAccessMode = ProducerAccessMode.Shared; public String getTopicName() { - return topicName; + return this.topicName; } public void setTopicName(String topicName) { @@ -388,7 +392,7 @@ public class PulsarProperties { } public String getProducerName() { - return producerName; + return this.producerName; } public void setProducerName(String producerName) { @@ -396,7 +400,7 @@ public class PulsarProperties { } public long getSendTimeoutMs() { - return sendTimeoutMs; + return this.sendTimeoutMs; } public void setSendTimeoutMs(long sendTimeoutMs) { @@ -404,7 +408,7 @@ public class PulsarProperties { } public boolean isBlockIfQueueFull() { - return blockIfQueueFull; + return this.blockIfQueueFull; } public void setBlockIfQueueFull(boolean blockIfQueueFull) { @@ -569,7 +573,7 @@ public class PulsarProperties { private int requestTimeoutMs = 60000; - private long initialBackoffIntervalNanos = TimeUnit.MILLISECONDS.toNanos(100);; + private long initialBackoffIntervalNanos = TimeUnit.MILLISECONDS.toNanos(100); private long maxBackoffIntervalNanos = TimeUnit.SECONDS.toNanos(30); diff --git a/spring-pulsar-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/AbstractContainerBaseTests.java b/spring-pulsar-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/AbstractContainerBaseTests.java index cfcbfb07..f03c027f 100644 --- a/spring-pulsar-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/AbstractContainerBaseTests.java +++ b/spring-pulsar-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/AbstractContainerBaseTests.java @@ -1,9 +1,25 @@ +/* + * 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.autoconfigure; import org.testcontainers.containers.PulsarContainer; import org.testcontainers.utility.DockerImageName; -public class AbstractContainerBaseTests { +abstract class AbstractContainerBaseTests { static final DockerImageName PULSAR_IMAGE = DockerImageName.parse("apachepulsar/pulsar:2.10.0"); diff --git a/spring-pulsar-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarListenerTests.java b/spring-pulsar-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarListenerTests.java index 13ba9205..88fdf109 100644 --- a/spring-pulsar-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarListenerTests.java +++ b/spring-pulsar-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarListenerTests.java @@ -16,6 +16,8 @@ package org.springframework.pulsar.autoconfigure; +import static org.assertj.core.api.Assertions.assertThat; + import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -30,8 +32,6 @@ import org.springframework.context.annotation.Import; import org.springframework.pulsar.annotation.PulsarListener; import org.springframework.pulsar.core.PulsarTemplate; -import static org.assertj.core.api.AssertionsForClassTypes.assertThat; - /** * @author Soby Chacko */ diff --git a/spring-pulsar-sample-apps/src/main/java/app1/PulsarBootApp.java b/spring-pulsar-sample-apps/src/main/java/app1/PulsarBootApp.java index b4f35d53..0a5c6496 100644 --- a/spring-pulsar-sample-apps/src/main/java/app1/PulsarBootApp.java +++ b/spring-pulsar-sample-apps/src/main/java/app1/PulsarBootApp.java @@ -1,3 +1,19 @@ +/* + * 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 app1; import org.apache.pulsar.common.schema.SchemaType; @@ -33,7 +49,7 @@ public class PulsarBootApp { @PulsarListener(subscriptionName = "test-exclusive-sub-2", topics = "hello-pulsar-exclusive-2", schemaType = SchemaType.JSON) public void listen(Foo foo) { - System.out.println("Message received: " + foo); + //... } static class Foo { @@ -41,7 +57,7 @@ public class PulsarBootApp { String bar; public String getFoo() { - return foo; + return this.foo; } public void setFoo(String foo) { @@ -49,7 +65,7 @@ public class PulsarBootApp { } public String getBar() { - return bar; + return this.bar; } public void setBar(String bar) { @@ -59,8 +75,8 @@ public class PulsarBootApp { @Override public String toString() { return "Foo{" + - "foo='" + foo + '\'' + - ", bar='" + bar + '\'' + + "foo='" + this.foo + '\'' + + ", bar='" + this.bar + '\'' + '}'; } } diff --git a/spring-pulsar-sample-apps/src/main/java/app2/ProducerApp.java b/spring-pulsar-sample-apps/src/main/java/app2/ProducerApp.java index 4bcc2ecd..d1f29726 100644 --- a/spring-pulsar-sample-apps/src/main/java/app2/ProducerApp.java +++ b/spring-pulsar-sample-apps/src/main/java/app2/ProducerApp.java @@ -1,3 +1,19 @@ +/* + * 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 app2; import java.io.Serial; @@ -29,16 +45,13 @@ public class ProducerApp { pulsarTemplate.setDefaultTopicName("failover-demo-topic"); return args -> { for (int i = 0; i < 100; i++) { - System.out.println("current i: " + i); pulsarTemplate.sendAsync("hello john doex " + new Random().nextInt(), new FooRouter()); pulsarTemplate.sendAsync("hello alice doex " + new Random().nextInt(), new BarRouter()); if (i % 2 == 0) { pulsarTemplate.sendAsync("hello buzz doex " + new Random().nextInt(), new BuzzRouter()); } Thread.sleep(5_000); - System.out.println("------------------------"); } - System.exit(0); }; } diff --git a/spring-pulsar-sample-apps/src/main/java/app3/PulsarAppTry.java b/spring-pulsar-sample-apps/src/main/java/app3/PulsarAppTry.java deleted file mode 100644 index 98dd2792..00000000 --- a/spring-pulsar-sample-apps/src/main/java/app3/PulsarAppTry.java +++ /dev/null @@ -1,66 +0,0 @@ -package app3; - -import org.springframework.boot.SpringApplication; -import org.springframework.boot.autoconfigure.SpringBootApplication; - -@SpringBootApplication -public class PulsarAppTry { - - public static void main(String[] args) { - - SpringApplication.run(PulsarAppTry.class, args); - } - -// @Bean -// public PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient) { -// Map config = new HashMap<>(); -// config.put("topicName", "foo-1"); -// return new DefaultPulsarProducerFactory<>(pulsarClient, config); -// } -// -// @Bean -// public PulsarClientFactoryBean pulsarClientFactoryBean(PulsarClientConfiguration pulsarClientConfiguration) { -// return new PulsarClientFactoryBean(pulsarClientConfiguration); -// } -// -// @Bean -// public PulsarClientConfiguration pulsarClientConfiguration() { -// return new PulsarClientConfiguration(); -// } - -// @Bean -// public PulsarTemplate pulsarTemplate(PulsarProducerFactory pulsarProducerFactory) { -// return new PulsarTemplate<>(pulsarProducerFactory); -// } -// -// @Bean -// public PulsarConsumerFactory pulsarConsumerFactory(PulsarClient pulsarClient) { -// -// Map config = new HashMap<>(); -//// final HashSet strings = new HashSet<>(); -//// strings.add("foobar-012"); -//// config.put("topicNames", strings); -//// config.put("subscriptionName", "foobar-sb-012"); -// -// return new DefaultPulsarConsumerFactory<>(pulsarClient, config); -// } -// -// @Bean -// PulsarListenerContainerFactory pulsarListenerContainerFactory(PulsarConsumerFactory pulsarConsumerFactory) { -// final PulsarListenerContainerFactoryImpl pulsarListenerContainerFactory = new PulsarListenerContainerFactoryImpl<>(); -// pulsarListenerContainerFactory.setPulsarConsumerFactory(pulsarConsumerFactory); -// return pulsarListenerContainerFactory; -// } - -// @PulsarListener(subscriptionName = "hello-pulsar-listener", topics = "foo-1") -// public void listen(String foo) { -// System.out.println("Message Received: " + foo); -// } - -// @Configuration(proxyBeanMethods = false) -// @EnablePulsar -// static class EnablePulsarConfiguration { -// -// } - -} diff --git a/spring-pulsar-sample-apps/src/main/java/app4/FailoverConsumerApp.java b/spring-pulsar-sample-apps/src/main/java/app4/FailoverConsumerApp.java index 858a2e73..ea178001 100644 --- a/spring-pulsar-sample-apps/src/main/java/app4/FailoverConsumerApp.java +++ b/spring-pulsar-sample-apps/src/main/java/app4/FailoverConsumerApp.java @@ -1,3 +1,19 @@ +/* + * 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 app4; import java.io.Serial; @@ -16,7 +32,6 @@ import org.springframework.pulsar.core.PulsarTemplate; @SpringBootApplication public class FailoverConsumerApp { - public static void main(String[] args) { String[] args1 = new String[]{ // "--spring.pulsar.consumer.subscription-type=Failover", @@ -33,7 +48,6 @@ public class FailoverConsumerApp { pulsarTemplate.sendAsync("hello alice doe 1", new BarRouter()); pulsarTemplate.sendAsync("hello buzz doe 2", new BuzzRouter()); Thread.sleep(1_000); - System.out.println("------------------------"); } System.exit(0); }; @@ -41,17 +55,17 @@ public class FailoverConsumerApp { @PulsarListener(subscriptionName = "failover-subscription-demo", topics = "failover-demo-topic", subscriptionType = "failover") public void listen1(String foo) { - System.out.println("Message Received 1: " + foo); + //... } @PulsarListener(subscriptionName = "failover-subscription-demo", topics = "failover-demo-topic", subscriptionType = "failover") public void listen2(String foo) { - System.out.println("Message Received 2: " + foo); + //... } @PulsarListener(subscriptionName = "failover-subscription-demo", topics = "failover-demo-topic", subscriptionType = "failover") public void listen(String foo) { - System.out.println("Message Received 3: " + foo); + //... } static class FooRouter implements MessageRouter { diff --git a/spring-pulsar-sample-apps/src/main/java/app5/FailoverConsumer.java b/spring-pulsar-sample-apps/src/main/java/app5/FailoverConsumer.java index d85620ea..d5c5d976 100644 --- a/spring-pulsar-sample-apps/src/main/java/app5/FailoverConsumer.java +++ b/spring-pulsar-sample-apps/src/main/java/app5/FailoverConsumer.java @@ -1,3 +1,19 @@ +/* + * 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 app5; import org.springframework.boot.SpringApplication; @@ -18,7 +34,7 @@ public class FailoverConsumer { @PulsarListener(subscriptionName = "failover-subscription-demo", topics = "failover-demo-topic", subscriptionType = "shared") public void listen1(String foo) { - System.out.println("Message Received: " + foo); + //... } diff --git a/spring-pulsar-sample-apps/src/main/java/app6/Sender.java b/spring-pulsar-sample-apps/src/main/java/app6/Sender.java deleted file mode 100644 index 6571c7ff..00000000 --- a/spring-pulsar-sample-apps/src/main/java/app6/Sender.java +++ /dev/null @@ -1,71 +0,0 @@ -package app6; - -import java.nio.charset.StandardCharsets; -import java.util.HashMap; -import java.util.Map; - -import org.apache.pulsar.client.api.PulsarClient; -import org.apache.pulsar.client.api.PulsarClientException; - -import org.springframework.context.annotation.AnnotationConfigApplicationContext; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.pulsar.config.PulsarClientConfiguration; -import org.springframework.pulsar.config.PulsarClientFactoryBean; -import org.springframework.pulsar.core.DefaultPulsarProducerFactory; -import org.springframework.pulsar.core.PulsarProducerFactory; -import org.springframework.pulsar.core.PulsarTemplate; - -public class Sender { - - public static void main(String[] args) throws PulsarClientException { - AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Config.class); - context.getBean(Sender.class).send(); - System.exit(0); - } - - private final PulsarTemplate template; - - public Sender(PulsarTemplate template) { - this.template = template; - } - - public void send() throws PulsarClientException { - this.template.send("foobar-fuzzy".getBytes(StandardCharsets.UTF_8)); - } - -} - - -@Configuration -class Config { - - @Bean - public PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient) { - Map config = new HashMap<>(); - config.put("topicName", "foo-1"); - return new DefaultPulsarProducerFactory<>(pulsarClient, config); - } - - @Bean - public PulsarTemplate pulsarTemplate(PulsarProducerFactory pulsarProducerFactory) { - return new PulsarTemplate<>(pulsarProducerFactory); - } - - @Bean - public Sender sender(PulsarTemplate template) { - return new Sender(template); - } - - @Bean - public PulsarClientFactoryBean pulsarClientFactoryBean(PulsarClientConfiguration pulsarClientConfiguration) { - return new PulsarClientFactoryBean(pulsarClientConfiguration); - } - - @Bean - public PulsarClientConfiguration pulsarClientConfiguration() { - return new PulsarClientConfiguration(); - } - -} - diff --git a/spring-pulsar-sample-apps/src/main/java/app6/SenderString.java b/spring-pulsar-sample-apps/src/main/java/app6/SenderString.java deleted file mode 100644 index ee20cebf..00000000 --- a/spring-pulsar-sample-apps/src/main/java/app6/SenderString.java +++ /dev/null @@ -1,83 +0,0 @@ -package app6; - -import java.util.HashMap; -import java.util.Map; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ExecutionException; - -import org.apache.pulsar.client.api.MessageId; -import org.apache.pulsar.client.api.PulsarClient; -import org.apache.pulsar.client.api.PulsarClientException; - -import org.springframework.context.annotation.AnnotationConfigApplicationContext; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.pulsar.config.PulsarClientConfiguration; -import org.springframework.pulsar.config.PulsarClientFactoryBean; -import org.springframework.pulsar.core.DefaultPulsarProducerFactory; -import org.springframework.pulsar.core.PulsarProducerFactory; -import org.springframework.pulsar.core.PulsarTemplate; - -public class SenderString { - - public static void main(String[] args) throws PulsarClientException { - AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(ConfigString.class); - context.getBean(SenderString.class).send(); - System.exit(0); - } - - private final PulsarTemplate template; - - public SenderString(PulsarTemplate template) { - this.template = template; - } - - public void send() throws PulsarClientException { - final CompletableFuture future = this.template.sendAsync("hello john doe"); - future.thenAccept(m -> System.out.println("Got " + m)); - try { - future.get(); - } - catch (InterruptedException e) { - e.printStackTrace(); - } - catch (ExecutionException e) { - e.printStackTrace(); - } - } - -} - - -@Configuration -class ConfigString { - - @Bean - public PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient) { - Map config = new HashMap<>(); - config.put("topicName", "foo-1"); - return new DefaultPulsarProducerFactory<>(pulsarClient, config); - } - - @Bean - public PulsarClientFactoryBean pulsarClientFactoryBean(PulsarClientConfiguration pulsarClientConfiguration) { - return new PulsarClientFactoryBean(pulsarClientConfiguration); - } - - @Bean - public PulsarClientConfiguration pulsarClientConfiguration() { - return new PulsarClientConfiguration(); - } - - @Bean - public PulsarTemplate pulsarTemplate(PulsarProducerFactory pulsarProducerFactory) { - return new PulsarTemplate<>(pulsarProducerFactory); - } - - @Bean - public SenderString senderString(PulsarTemplate template) { - return new SenderString(template); - } - -} - diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/PulsarException.java b/spring-pulsar/src/main/java/org/springframework/pulsar/PulsarException.java index f8f73134..e6302754 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/PulsarException.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/PulsarException.java @@ -19,8 +19,11 @@ package org.springframework.pulsar; import org.springframework.core.NestedRuntimeException; /** + * Spring Pulsar specific {@link NestedRuntimeException} implementation. + * * @author Soby Chacko */ +@SuppressWarnings("serial") public class PulsarException extends NestedRuntimeException { public PulsarException(String msg) { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarBootstrapConfiguration.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarBootstrapConfiguration.java index 0ed4e146..bde6dc7e 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarBootstrapConfiguration.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarBootstrapConfiguration.java @@ -20,7 +20,6 @@ import org.springframework.beans.factory.support.BeanDefinitionRegistry; import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.context.annotation.ImportBeanDefinitionRegistrar; import org.springframework.core.type.AnnotationMetadata; -import org.springframework.pulsar.annotation.PulsarListenerAnnotationBeanPostProcessor; import org.springframework.pulsar.config.PulsarListenerConfigUtils; import org.springframework.pulsar.config.PulsarListenerEndpointRegistry; @@ -30,7 +29,7 @@ import org.springframework.pulsar.config.PulsarListenerEndpointRegistry; * a default {@link PulsarListenerEndpointRegistry}. * *

This configuration class is automatically imported when using the @{@link EnablePulsar} - * annotation. + * annotation. * * @author Soby Chacko * @@ -55,4 +54,4 @@ public class PulsarBootstrapConfiguration implements ImportBeanDefinitionRegistr } } -} \ No newline at end of file +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java index 011f69be..05610016 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java @@ -22,7 +22,6 @@ import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; -import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.common.schema.SchemaType; import org.springframework.messaging.handler.annotation.MessageMapping; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java index 911cf4be..0f0e706e 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java @@ -39,9 +39,7 @@ import java.util.function.BiFunction; import java.util.stream.Collectors; import org.apache.commons.logging.LogFactory; -import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.SubscriptionType; -import org.apache.pulsar.common.schema.SchemaType; import org.springframework.aop.framework.Advised; import org.springframework.aop.support.AopUtils; @@ -124,6 +122,9 @@ public class PulsarListenerAnnotationBeanPostProcessor implements BeanPost private final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); + /** + * The bean name of the default {@link org.springframework.pulsar.config.PulsarListenerContainerFactory}. + */ public static final String DEFAULT_PULSAR_LISTENER_CONTAINER_FACTORY_BEAN_NAME = "pulsarListenerContainerFactory"; private static final String THE_LEFT = "The ["; @@ -289,7 +290,7 @@ public class PulsarListenerAnnotationBeanPostProcessor implements BeanPost } protected void processListener(MethodPulsarListenerEndpoint endpoint, PulsarListener PulsarListener, - Object bean, String beanName, String[] topics) { + Object bean, String beanName, String[] topics) { processPulsarListenerAnnotation(endpoint, PulsarListener, bean, topics); @@ -302,7 +303,7 @@ public class PulsarListenerAnnotationBeanPostProcessor implements BeanPost @Nullable private PulsarListenerContainerFactory resolveContainerFactory(PulsarListener PulsarListener, - Object factoryTarget, String beanName) { + Object factoryTarget, String beanName) { String containerFactory = PulsarListener.containerFactory(); if (!StringUtils.hasText(containerFactory)) { @@ -344,7 +345,7 @@ public class PulsarListenerAnnotationBeanPostProcessor implements BeanPost } private void processPulsarListenerAnnotation(MethodPulsarListenerEndpoint endpoint, - PulsarListener pulsarListener, Object bean, String[] topics) { + PulsarListener pulsarListener, Object bean, String[] topics) { endpoint.setBean(bean); endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerConfigurer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerConfigurer.java index 59ad2161..636f92f8 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerConfigurer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerConfigurer.java @@ -19,6 +19,8 @@ package org.springframework.pulsar.annotation; import org.springframework.pulsar.config.PulsarListenerEndpointRegistrar; /** + * Allow custom configuration on PulsarListener endpoint registry. + * * @author Soby Chacko */ public interface PulsarListenerConfigurer { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListeners.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListeners.java index ed9a3883..e9f3b0c4 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListeners.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListeners.java @@ -22,9 +22,9 @@ import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; -import org.springframework.pulsar.annotation.PulsarListener; - /** + * Container annotation for aggregating several {@link PulsarListener} annotations. + * * @author Soby Chacko */ @Target({ ElementType.TYPE, ElementType.METHOD, ElementType.ANNOTATION_TYPE }) @@ -34,4 +34,4 @@ public @interface PulsarListeners { PulsarListener[] value(); -} \ No newline at end of file +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java index b5fc8bb3..3a4a7ec2 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java @@ -19,7 +19,6 @@ package org.springframework.pulsar.config; import org.apache.commons.logging.LogFactory; import org.apache.pulsar.client.api.Schema; -import org.springframework.beans.BeanUtils; import org.springframework.beans.BeansException; import org.springframework.beans.factory.InitializingBean; import org.springframework.context.ApplicationContext; @@ -27,13 +26,18 @@ import org.springframework.context.ApplicationContextAware; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.core.log.LogAccessor; +import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.listener.AbstractPulsarMessageListenerContainer; +import org.springframework.pulsar.listener.PulsarContainerProperties; import org.springframework.pulsar.support.JavaUtils; import org.springframework.pulsar.support.MessageConverter; -import org.springframework.pulsar.core.PulsarConsumerFactory; -import org.springframework.pulsar.listener.PulsarContainerProperties; /** + * Base {@link PulsarListenerContainerFactory} implementation. + * + * @param the {@link AbstractPulsarMessageListenerContainer} implementation type. + * @param Message payload type. + * * @author Soby Chacko */ public abstract class AbstractPulsarListenerContainerFactory, T> diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerEndpoint.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerEndpoint.java index a6bbaf8b..1eb03260 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerEndpoint.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerEndpoint.java @@ -37,12 +37,16 @@ import org.springframework.context.expression.BeanFactoryResolver; import org.springframework.core.log.LogAccessor; import org.springframework.expression.BeanResolver; import org.springframework.lang.Nullable; -import org.springframework.pulsar.support.MessageConverter; import org.springframework.pulsar.listener.PulsarMessageListenerContainer; import org.springframework.pulsar.listener.adapter.PulsarMessagingMessageListenerAdapter; +import org.springframework.pulsar.support.MessageConverter; import org.springframework.util.Assert; /** + * Base implementation for {@link PulsarListenerEndpoint}. + * + * @param Message payload type. + * * @author Soby Chacko */ public abstract class AbstractPulsarListenerEndpoint implements PulsarListenerEndpoint, BeanFactoryAware, InitializingBean { @@ -152,14 +156,14 @@ public abstract class AbstractPulsarListenerEndpoint implements PulsarListene @Override public void setupListenerContainer(PulsarMessageListenerContainer listenerContainer, - @Nullable MessageConverter messageConverter) { + @Nullable MessageConverter messageConverter) { setupMessageListener(listenerContainer, messageConverter); } @SuppressWarnings("unchecked") private void setupMessageListener(PulsarMessageListenerContainer container, - @Nullable MessageConverter messageConverter) { + @Nullable MessageConverter messageConverter) { PulsarMessagingMessageListenerAdapter adapter = createMessageListener(container, messageConverter); Object messageListener = adapter; @@ -170,7 +174,7 @@ public abstract class AbstractPulsarListenerEndpoint implements PulsarListene } protected abstract PulsarMessagingMessageListenerAdapter createMessageListener(PulsarMessageListenerContainer container, - @Nullable MessageConverter messageConverter); + @Nullable MessageConverter messageConverter); public void setConsumerProperties(Properties consumerProperties) { this.consumerProperties = consumerProperties; @@ -192,7 +196,7 @@ public abstract class AbstractPulsarListenerEndpoint implements PulsarListene public SubscriptionType getSubscriptionType() { - return subscriptionType; + return this.subscriptionType; } public void setSubscriptionType(SubscriptionType subscriptionType) { @@ -200,7 +204,7 @@ public abstract class AbstractPulsarListenerEndpoint implements PulsarListene } public SchemaType getSchemaType() { - return schemaType; + return this.schemaType; } public void setSchemaType(SchemaType schemaType) { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java index 67222ff7..ba0813e8 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java @@ -50,6 +50,11 @@ import org.springframework.pulsar.support.converter.PulsarRecordMessageConverter import org.springframework.util.Assert; /** + * A {@link PulsarListenerEndpoint} providing the method to invoke to process + * an incoming message for this endpoint. + * + * @param Message payload type + * * @author Soby Chacko */ public class MethodPulsarListenerEndpoint extends AbstractPulsarListenerEndpoint { @@ -91,7 +96,7 @@ public class MethodPulsarListenerEndpoint extends AbstractPulsarListenerEndpo @Override protected PulsarMessagingMessageListenerAdapter createMessageListener(PulsarMessageListenerContainer container, - @Nullable MessageConverter messageConverter) { + @Nullable MessageConverter messageConverter) { Assert.state(this.messageHandlerMethodFactory != null, "Could not create message listener - MessageHandlerMethodFactory not set"); PulsarMessagingMessageListenerAdapter messageListener = createMessageListenerInstance(messageConverter); @@ -226,5 +231,4 @@ public class MethodPulsarListenerEndpoint extends AbstractPulsarListenerEndpo this.messagingConverter = messagingConverter; } - } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarClientConfiguration.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarClientConfiguration.java index 4251e29c..1f129908 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarClientConfiguration.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarClientConfiguration.java @@ -22,6 +22,8 @@ import java.util.Map; import org.springframework.util.Assert; /** + * Configuration for the Pulsar client. + * * @author Soby Chacko */ public class PulsarClientConfiguration { @@ -39,6 +41,6 @@ public class PulsarClientConfiguration { } public Map getConfigs() { - return configs; + return this.configs; } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarClientFactoryBean.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarClientFactoryBean.java index 20006f6c..03357ee5 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarClientFactoryBean.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarClientFactoryBean.java @@ -19,10 +19,10 @@ package org.springframework.pulsar.config; import org.apache.pulsar.client.api.PulsarClient; import org.springframework.beans.factory.config.AbstractFactoryBean; -import org.springframework.lang.Nullable; -import org.springframework.pulsar.config.PulsarClientConfiguration; /** + * {@link org.springframework.beans.factory.FactoryBean} implementation for the {@link PulsarClient}. + * * @author Soby Chacko */ public class PulsarClientFactoryBean extends AbstractFactoryBean { @@ -48,7 +48,7 @@ public class PulsarClientFactoryBean extends AbstractFactoryBean { @Override protected void destroyInstance(PulsarClient instance) throws Exception { if (instance != null) { - System.out.printf("CLOSING THE CLIENT"); + this.logger.info("Closing the client: " + instance); instance.close(); } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerConfigUtils.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerConfigUtils.java index 3acaec4f..9be83f4b 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerConfigUtils.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerConfigUtils.java @@ -17,6 +17,8 @@ package org.springframework.pulsar.config; /** + * Configuration constants for internal sharing across subpackages. + * * @author Soby Chacko */ public abstract class PulsarListenerConfigUtils { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerContainerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerContainerFactory.java index a6b594e0..ecd8447d 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerContainerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerContainerFactory.java @@ -19,6 +19,10 @@ package org.springframework.pulsar.config; import org.springframework.pulsar.listener.PulsarMessageListenerContainer; /** + * Factory for Pulsar message listener containers. + * + * @param message listener container type. + * * @author Soby Chacko */ public interface PulsarListenerContainerFactory { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerContainerFactoryImpl.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerContainerFactoryImpl.java index 822304d1..9067fdd2 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerContainerFactoryImpl.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerContainerFactoryImpl.java @@ -26,6 +26,11 @@ import org.springframework.pulsar.listener.PulsarContainerProperties; import org.springframework.util.StringUtils; /** + * Concrete implementation for {@link PulsarListenerContainerFactory}. + * + * @param container implementation type. + * @param message type in the listener. + * * @author Soby Chacko */ public class PulsarListenerContainerFactoryImpl extends AbstractPulsarListenerContainerFactory, T> { @@ -61,7 +66,7 @@ public class PulsarListenerContainerFactoryImpl extends AbstractPulsarList @Override protected void initializeContainer(DefaultPulsarMessageListenerContainer instance, - PulsarListenerEndpoint endpoint) { + PulsarListenerEndpoint endpoint) { super.initializeContainer(instance, endpoint); } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpoint.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpoint.java index af7499bb..00410113 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpoint.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpoint.java @@ -22,10 +22,14 @@ import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.common.schema.SchemaType; import org.springframework.lang.Nullable; -import org.springframework.pulsar.support.MessageConverter; import org.springframework.pulsar.listener.PulsarMessageListenerContainer; +import org.springframework.pulsar.support.MessageConverter; /** + * Model for a Pulsar listener endpoint. Can be used against a + * {@link org.springframework.pulsar.annotation.PulsarListenerConfigurer} + * to register endpoints programmatically. + * * @author Soby Chacko */ public interface PulsarListenerEndpoint { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointAdapter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointAdapter.java index 63fb17a5..b700b472 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointAdapter.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointAdapter.java @@ -22,10 +22,12 @@ import java.util.Collections; import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.common.schema.SchemaType; -import org.springframework.pulsar.support.MessageConverter; import org.springframework.pulsar.listener.PulsarMessageListenerContainer; +import org.springframework.pulsar.support.MessageConverter; /** + * Adapter to avoid having to implement all methods. + * * @author Soby Chacko */ public class PulsarListenerEndpointAdapter implements PulsarListenerEndpoint { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointRegistrar.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointRegistrar.java index abee3b9e..0c8eafb9 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointRegistrar.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointRegistrar.java @@ -31,27 +31,31 @@ import org.springframework.util.Assert; import org.springframework.validation.Validator; /** + * Helper bean for registering {@link PulsarListenerEndpoint} with + * a {@link PulsarListenerEndpointRegistry}. + * * @author Soby Chacko */ public class PulsarListenerEndpointRegistrar implements BeanFactoryAware, InitializingBean { private final List endpointDescriptors = new ArrayList<>(); - private PulsarListenerEndpointRegistry endpointRegistry; + private PulsarListenerEndpointRegistry endpointRegistry; private List customMethodArgumentResolvers = new ArrayList<>(); private Validator validator; private MessageHandlerMethodFactory messageHandlerMethodFactory; + private PulsarListenerContainerFactory containerFactory; + private String containerFactoryBeanName; + private BeanFactory beanFactory; - private boolean startImmediately; - public void setEndpointRegistry(PulsarListenerEndpointRegistry endpointRegistry) { this.endpointRegistry = endpointRegistry; } @@ -120,7 +124,6 @@ public class PulsarListenerEndpointRegistrar implements BeanFactoryAware, Initia } } - private PulsarListenerContainerFactory resolveContainerFactory(PulsarListenerEndpointDescriptor descriptor) { if (descriptor.containerFactory != null) { return descriptor.containerFactory; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointRegistry.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointRegistry.java index 844bd3eb..ef87511b 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointRegistry.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointRegistry.java @@ -43,6 +43,18 @@ import org.springframework.pulsar.support.EndpointHandlerMethod; import org.springframework.util.Assert; /** + * Creates the necessary {@link PulsarMessageListenerContainer} instances for the + * registered {@linkplain PulsarListenerEndpoint endpoints}. Also manages the + * lifecycle of the listener containers, in particular within the lifecycle + * of the application context. + * + *

Contrary to {@link PulsarMessageListenerContainer}s created manually, listener + * containers managed by registry are not beans in the application context and + * are not candidates for autowiring. Use {@link #getListenerContainers()} if + * you need to access this registry's listener containers for management purposes. + * If you need to access to a specific message listener container, use + * {@link #getListenerContainer(String)} with the id of the endpoint. + * * @author Soby Chacko */ public class PulsarListenerEndpointRegistry implements PulsarListenerContainerRegistry, DisposableBean, SmartLifecycle, @@ -96,7 +108,7 @@ public class PulsarListenerEndpointRegistry implements PulsarListenerContainerRe } public void registerListenerContainer(PulsarListenerEndpoint endpoint, PulsarListenerContainerFactory factory, - boolean startImmediately) { + boolean startImmediately) { Assert.notNull(endpoint, "Endpoint must not be null"); Assert.notNull(factory, "Factory must not be null"); @@ -115,7 +127,7 @@ public class PulsarListenerEndpointRegistry implements PulsarListenerContainerRe } protected PulsarMessageListenerContainer createListenerContainer(PulsarListenerEndpoint endpoint, - PulsarListenerContainerFactory factory) { + PulsarListenerContainerFactory factory) { if (endpoint instanceof MethodPulsarListenerEndpoint) { MethodPulsarListenerEndpoint mkle = (MethodPulsarListenerEndpoint) endpoint; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java index d65727ab..00574430 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java @@ -31,6 +31,10 @@ import org.apache.pulsar.client.api.Schema; import org.springframework.util.CollectionUtils; /** + * Default implementation for {@link PulsarConsumerFactory}. + * + * @param underlying payload type for the consumer. + * * @author Soby Chacko */ public class DefaultPulsarConsumerFactory implements PulsarConsumerFactory { @@ -60,7 +64,7 @@ public class DefaultPulsarConsumerFactory implements PulsarConsumerFactory consumerBuilder.loadConf(properties); } Consumer consumer = consumerBuilder.subscribe(); - consumers.add(consumer); + this.consumers.add(consumer); return consumer; } @@ -77,12 +81,12 @@ public class DefaultPulsarConsumerFactory implements PulsarConsumerFactory consumerBuilder.batchReceivePolicy(batchReceivePolicy); Consumer consumer = consumerBuilder.subscribe(); - consumers.add(consumer); + this.consumers.add(consumer); return consumer; } public Map getConsumerConfig() { - return consumerConfig; + return this.consumerConfig; } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java index 218e9291..e5733485 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java @@ -32,6 +32,10 @@ import org.springframework.core.log.LogAccessor; import org.springframework.util.CollectionUtils; /** + * Default implementation for {@link PulsarProducerFactory}. + * + * @param producer type. + * * @author Soby Chacko */ public class DefaultPulsarProducerFactory implements PulsarProducerFactory, DisposableBean { @@ -60,7 +64,7 @@ public class DefaultPulsarProducerFactory implements PulsarProducerFactory producerBuilder.loadConf(this.producerConfig); } this.producer = producerBuilder.create(); - return producer; + return this.producer; } @Override @@ -73,12 +77,12 @@ public class DefaultPulsarProducerFactory implements PulsarProducerFactory } producerBuilder.messageRouter(messageRouter); this.producer = producerBuilder.create(); - return producer; + return this.producer; } @Override public Map getProducerConfig() { - return producerConfig; + return this.producerConfig; } @Override diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarClientProperties.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarClientProperties.java deleted file mode 100644 index 3d0024fb..00000000 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarClientProperties.java +++ /dev/null @@ -1,245 +0,0 @@ -/* - * 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.core; - -import java.net.SocketAddress; - -/** - * @author Soby Chacko - */ -public class PulsarClientProperties { - - private String serviceUrl; - - private String authPluginClassName; - - private String authParams; - - private long operationTimeoutMs; - - private long statsIntervalSeconds; - - private int numIoThreads; - - private boolean useTcpNoDelay; - - private boolean useTls; - - private String tlsTrustCertsFilePath; - - private boolean tlsAllowInsecureConnection; - - private boolean tlsHostnameVerificationEnable; - - private int concurrentLookupRequest; - - private int maxLookupRequest; - - private int maxNumberOfRejectedRequestPerConnection; - - private int keepAliveIntervalSeconds; - - private int connectionTimeoutMs; - - private int requestTimeoutMs; - - private int defaultBackoffIntervalNanos; - - private long maxBackoffIntervalNanos; - - private SocketAddress socks5ProxyAddress; - - private String socks5ProxyUsername; - - private String socks5ProxyPassword; - - public String getServiceUrl() { - return serviceUrl; - } - - public void setServiceUrl(String serviceUrl) { - this.serviceUrl = serviceUrl; - } - - public String getAuthPluginClassName() { - return authPluginClassName; - } - - public void setAuthPluginClassName(String authPluginClassName) { - this.authPluginClassName = authPluginClassName; - } - - public String getAuthParams() { - return authParams; - } - - public void setAuthParams(String authParams) { - this.authParams = authParams; - } - - public long getOperationTimeoutMs() { - return operationTimeoutMs; - } - - public void setOperationTimeoutMs(long operationTimeoutMs) { - this.operationTimeoutMs = operationTimeoutMs; - } - - public long getStatsIntervalSeconds() { - return statsIntervalSeconds; - } - - public void setStatsIntervalSeconds(long statsIntervalSeconds) { - this.statsIntervalSeconds = statsIntervalSeconds; - } - - public int getNumIoThreads() { - return numIoThreads; - } - - public void setNumIoThreads(int numIoThreads) { - this.numIoThreads = numIoThreads; - } - - public boolean isUseTcpNoDelay() { - return useTcpNoDelay; - } - - public void setUseTcpNoDelay(boolean useTcpNoDelay) { - this.useTcpNoDelay = useTcpNoDelay; - } - - public boolean isUseTls() { - return useTls; - } - - public void setUseTls(boolean useTls) { - this.useTls = useTls; - } - - public String getTlsTrustCertsFilePath() { - return tlsTrustCertsFilePath; - } - - public void setTlsTrustCertsFilePath(String tlsTrustCertsFilePath) { - this.tlsTrustCertsFilePath = tlsTrustCertsFilePath; - } - - public boolean isTlsAllowInsecureConnection() { - return tlsAllowInsecureConnection; - } - - public void setTlsAllowInsecureConnection(boolean tlsAllowInsecureConnection) { - this.tlsAllowInsecureConnection = tlsAllowInsecureConnection; - } - - public boolean isTlsHostnameVerificationEnable() { - return tlsHostnameVerificationEnable; - } - - public void setTlsHostnameVerificationEnable(boolean tlsHostnameVerificationEnable) { - this.tlsHostnameVerificationEnable = tlsHostnameVerificationEnable; - } - - public int getConcurrentLookupRequest() { - return concurrentLookupRequest; - } - - public void setConcurrentLookupRequest(int concurrentLookupRequest) { - this.concurrentLookupRequest = concurrentLookupRequest; - } - - public int getMaxLookupRequest() { - return maxLookupRequest; - } - - public void setMaxLookupRequest(int maxLookupRequest) { - this.maxLookupRequest = maxLookupRequest; - } - - public int getMaxNumberOfRejectedRequestPerConnection() { - return maxNumberOfRejectedRequestPerConnection; - } - - public void setMaxNumberOfRejectedRequestPerConnection(int maxNumberOfRejectedRequestPerConnection) { - this.maxNumberOfRejectedRequestPerConnection = maxNumberOfRejectedRequestPerConnection; - } - - public int getKeepAliveIntervalSeconds() { - return keepAliveIntervalSeconds; - } - - public void setKeepAliveIntervalSeconds(int keepAliveIntervalSeconds) { - this.keepAliveIntervalSeconds = keepAliveIntervalSeconds; - } - - public int getConnectionTimeoutMs() { - return connectionTimeoutMs; - } - - public void setConnectionTimeoutMs(int connectionTimeoutMs) { - this.connectionTimeoutMs = connectionTimeoutMs; - } - - public int getRequestTimeoutMs() { - return requestTimeoutMs; - } - - public void setRequestTimeoutMs(int requestTimeoutMs) { - this.requestTimeoutMs = requestTimeoutMs; - } - - public int getDefaultBackoffIntervalNanos() { - return defaultBackoffIntervalNanos; - } - - public void setDefaultBackoffIntervalNanos(int defaultBackoffIntervalNanos) { - this.defaultBackoffIntervalNanos = defaultBackoffIntervalNanos; - } - - public long getMaxBackoffIntervalNanos() { - return maxBackoffIntervalNanos; - } - - public void setMaxBackoffIntervalNanos(long maxBackoffIntervalNanos) { - this.maxBackoffIntervalNanos = maxBackoffIntervalNanos; - } - - public SocketAddress getSocks5ProxyAddress() { - return socks5ProxyAddress; - } - - public void setSocks5ProxyAddress(SocketAddress socks5ProxyAddress) { - this.socks5ProxyAddress = socks5ProxyAddress; - } - - public String getSocks5ProxyUsername() { - return socks5ProxyUsername; - } - - public void setSocks5ProxyUsername(String socks5ProxyUsername) { - this.socks5ProxyUsername = socks5ProxyUsername; - } - - public String getSocks5ProxyPassword() { - return socks5ProxyPassword; - } - - public void setSocks5ProxyPassword(String socks5ProxyPassword) { - this.socks5ProxyPassword = socks5ProxyPassword; - } -} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarConsumerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarConsumerFactory.java index 8003e18f..308e8aa8 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarConsumerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarConsumerFactory.java @@ -24,6 +24,10 @@ import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; /** + * Pulsar consumer factory interface. + * + * @param payload type for the consumer. + * * @author Soby Chacko */ public interface PulsarConsumerFactory { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java index fc4730ab..292a5e61 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java @@ -20,11 +20,14 @@ import java.util.Map; import org.apache.pulsar.client.api.MessageRouter; import org.apache.pulsar.client.api.Producer; -import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; /** + * Pulsar producer factory interface. + * + * @param producer payload type. + * * @author Soby Chacko */ public interface PulsarProducerFactory { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java index 93dc8cc9..acfb2fca 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java @@ -28,6 +28,10 @@ import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; /** + * Template implementation for publishing to Pulsar topics. + * + * @param message type. + * * @author Soby Chacko */ public class PulsarTemplate { @@ -47,10 +51,10 @@ public class PulsarTemplate { public MessageId send(T message) throws PulsarClientException { final Schema schema = this.schema != null ? this.schema : SchemaUtils.getSchema(message); final SchemaTopic schemaTopic = getSchemaTopic(schema, this.pulsarProducerFactory, null); - Producer producer = producerCache.get(schemaTopic); + Producer producer = this.producerCache.get(schemaTopic); if (producer == null) { producer = this.pulsarProducerFactory.createProducer(schema); - producerCache.put(schemaTopic, producer); + this.producerCache.put(schemaTopic, producer); } return producer.send(message); } @@ -58,10 +62,10 @@ public class PulsarTemplate { public CompletableFuture sendAsync(T message) throws PulsarClientException { final Schema schema = this.schema != null ? this.schema : SchemaUtils.getSchema(message); final SchemaTopic schemaTopic = getSchemaTopic(schema, this.pulsarProducerFactory, null); - Producer producer = producerCache.get(schemaTopic); + Producer producer = this.producerCache.get(schemaTopic); if (producer == null) { producer = this.pulsarProducerFactory.createProducer(schema); - producerCache.put(schemaTopic, producer); + this.producerCache.put(schemaTopic, producer); } return producer.sendAsync(message); } @@ -69,10 +73,10 @@ public class PulsarTemplate { public CompletableFuture sendAsync(T message, MessageRouter messageRouter) throws PulsarClientException { final Schema schema = this.schema != null ? this.schema : SchemaUtils.getSchema(message); final SchemaTopic schemaTopic = getSchemaTopic(schema, this.pulsarProducerFactory, messageRouter); - Producer producer = producerCache.get(schemaTopic); + Producer producer = this.producerCache.get(schemaTopic); if (producer == null) { producer = this.pulsarProducerFactory.createProducer(schema, messageRouter); - producerCache.put(schemaTopic, producer); + this.producerCache.put(schemaTopic, producer); } return producer.sendAsync(message); } @@ -87,7 +91,7 @@ public class PulsarTemplate { } public Schema getSchema() { - return schema; + return this.schema; } public void setSchema(Schema schema) { @@ -100,7 +104,7 @@ public class PulsarTemplate { final String topicName; final MessageRouter messageRouter; - public SchemaTopic(Schema schema, String topicName, MessageRouter messageRouter) { + SchemaTopic(Schema schema, String topicName, MessageRouter messageRouter) { this.schema = schema; this.topicName = topicName; this.messageRouter = messageRouter; @@ -108,12 +112,16 @@ public class PulsarTemplate { @Override public boolean equals(Object o) { - if (this == o) return true; - if (o == null || getClass() != o.getClass()) return false; + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } @SuppressWarnings("unchecked") SchemaTopic that = (SchemaTopic) o; if (this.messageRouter == null && that.messageRouter == null) { - return schema.equals(that.schema) && topicName.equals(that.topicName); + return this.schema.equals(that.schema) && this.topicName.equals(that.topicName); } else if (this.messageRouter == null) { return false; @@ -121,12 +129,12 @@ public class PulsarTemplate { else if (that.messageRouter == null) { return false; } - return schema.equals(that.schema) && topicName.equals(that.topicName) && messageRouter.equals(that.messageRouter); + return this.schema.equals(that.schema) && this.topicName.equals(that.topicName) && this.messageRouter.equals(that.messageRouter); } @Override public int hashCode() { - return Objects.hash(schema, topicName, messageRouter); + return Objects.hash(this.schema, this.topicName, this.messageRouter); } } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaUtils.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaUtils.java index 4eecc1bb..12e95d74 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaUtils.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaUtils.java @@ -1,7 +1,7 @@ /* * Copyright 2022 the original author or authors. * - * Licensed under the Apache License, Version 2.0 (the "License"; + * 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 * @@ -20,9 +20,15 @@ import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.impl.schema.JSONSchema; /** + * Utility class for Pulsar schema inference. + * * @author Soby Chacko */ -public class SchemaUtils { +public final class SchemaUtils { + + private SchemaUtils() { + + } @SuppressWarnings("unchecked") public static Schema getSchema(T message) { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/event/ConsumerFailedToStartEvent.java b/spring-pulsar/src/main/java/org/springframework/pulsar/event/ConsumerFailedToStartEvent.java index a618e150..342e7f7e 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/event/ConsumerFailedToStartEvent.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/event/ConsumerFailedToStartEvent.java @@ -17,6 +17,8 @@ package org.springframework.pulsar.event; /** + * Event to publish when the consumer is failed to start. + * * @author Soby Chacko */ public class ConsumerFailedToStartEvent extends PulsarEvent { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/event/ConsumerStartedEvent.java b/spring-pulsar/src/main/java/org/springframework/pulsar/event/ConsumerStartedEvent.java index f6f9c6e0..f85ef8e8 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/event/ConsumerStartedEvent.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/event/ConsumerStartedEvent.java @@ -17,6 +17,8 @@ package org.springframework.pulsar.event; /** + * Event to publish when the consumer is started. + * * @author Soby Chacko */ public class ConsumerStartedEvent extends PulsarEvent { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/event/ConsumerStartingEvent.java b/spring-pulsar/src/main/java/org/springframework/pulsar/event/ConsumerStartingEvent.java index 1f4d1e56..7c16ec84 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/event/ConsumerStartingEvent.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/event/ConsumerStartingEvent.java @@ -17,6 +17,8 @@ package org.springframework.pulsar.event; /** + * Event to publish while the consumer is starting. + * * @author Soby Chacko */ public class ConsumerStartingEvent extends PulsarEvent { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/event/PulsarEvent.java b/spring-pulsar/src/main/java/org/springframework/pulsar/event/PulsarEvent.java index 827b2dcc..babeed19 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/event/PulsarEvent.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/event/PulsarEvent.java @@ -20,6 +20,8 @@ import org.springframework.context.ApplicationEvent; import org.springframework.util.Assert; /** + * Base class for events. + * * @author Soby Chacko */ public class PulsarEvent extends ApplicationEvent { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java index 3f9e13fa..b4ce7b6a 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java @@ -29,6 +29,10 @@ import org.springframework.lang.Nullable; import org.springframework.pulsar.core.PulsarConsumerFactory; /** + * Base implementation for the {@link PulsarMessageListenerContainer}. + * + * @param message type. + * * @author Soby Chacko */ public abstract class AbstractPulsarMessageListenerContainer @@ -49,9 +53,10 @@ public abstract class AbstractPulsarMessageListenerContainer private int phase; @SuppressWarnings("unchecked") - protected AbstractPulsarMessageListenerContainer(PulsarConsumerFactory pulsarConsumerFactory,PulsarContainerProperties pulsarContainerProperties) { + protected AbstractPulsarMessageListenerContainer(PulsarConsumerFactory pulsarConsumerFactory, + PulsarContainerProperties pulsarContainerProperties) { this.pulsarContainerProperties = pulsarContainerProperties; - this.pulsarConsumerFactory = (PulsarConsumerFactory)pulsarConsumerFactory; + this.pulsarConsumerFactory = (PulsarConsumerFactory) pulsarConsumerFactory; } @@ -99,11 +104,11 @@ public abstract class AbstractPulsarMessageListenerContainer } public PulsarContainerProperties getPulsarContainerProperties() { - return pulsarContainerProperties; + return this.pulsarContainerProperties; } public PulsarConsumerFactory getPulsarConsumerFactory() { - return pulsarConsumerFactory; + return this.pulsarConsumerFactory; } @Override diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java index 5f05c10e..e2e026de 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java @@ -45,6 +45,10 @@ import org.springframework.util.StringUtils; import org.springframework.util.concurrent.ListenableFuture; /** + * Default implementation for {@link PulsarMessageListenerContainer}. + * + * @param message type. + * * @author Soby Chacko */ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMessageListenerContainer { @@ -243,11 +247,11 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess Messages messages = null; try { // Always receive messages in batch mode. - messages = consumer.batchReceive(); + messages = this.consumer.batchReceive(); // TODO Async receive - both record and batch. if (this.containerProperties.isBatchListener()) { try { - this.batchMessageHandler.received(consumer, messages); + this.batchMessageHandler.received(this.consumer, messages); this.consumer.acknowledge(messages); } catch (Exception e) { @@ -258,7 +262,7 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess else { for (Message message : messages) { try { - this.listener.received(consumer, message); + this.listener.received(this.consumer, message); if (this.containerProperties.getAckMode() != PulsarContainerProperties.AckMode.MANUAL) { this.consumer.acknowledge(message); } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarConsumerProperties.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarConsumerProperties.java deleted file mode 100644 index d515a46e..00000000 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarConsumerProperties.java +++ /dev/null @@ -1,309 +0,0 @@ -///* -// * 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.listener; -// -//import java.util.SortedMap; -//import java.util.regex.Pattern; -// -//import org.apache.pulsar.client.api.ConsumerCryptoFailureAction; -//import org.apache.pulsar.client.api.DeadLetterPolicy; -//import org.apache.pulsar.client.api.RedeliveryBackoff; -//import org.apache.pulsar.client.api.RegexSubscriptionMode; -//import org.apache.pulsar.client.api.SubscriptionInitialPosition; -//import org.apache.pulsar.client.api.SubscriptionType; -// -///** -// * @author Soby Chacko -// */ -//public class PulsarConsumerProperties { -// -// /** -// * Topic names. -// */ -// private String[] topics; -// -// /** -// * Topic pattern. -// */ -// private Pattern topicsPattern; -// -// private String subscriptionName; -// -// private SubscriptionType subscriptionType; -// -// private int receiveQueueSize; -// -// private long acknowledgementsGroupTimeMicros; -// -// private long negativeAckRedeliveryDelayMicros; -// -// private int maxTotalReceiverQueueSizeAcrossPartitions; -// -// private String consumerName; -// -// private long ackTimeoutMillis; -// -// private long tickDurationMillis; -// -// private int priorityLevel; -// -// private ConsumerCryptoFailureAction cryptoFailureAction; -// -// private SortedMap properties; -// -// private boolean readCompacted; -// -// private SubscriptionInitialPosition subscriptionInitialPosition; -// -// private int patternAutoDiscoveryPeriod; -// -// private RegexSubscriptionMode regexSubscriptionMode; -// -// private DeadLetterPolicy deadLetterPolicy; -// -// private boolean autoUpdatePartitions; -// -// private boolean replicateSubscriptionState; -// -// private RedeliveryBackoff negativeAckRedeliveryBackoff; -// private RedeliveryBackoff ackTimeoutRedeliveryBackoff; -// private boolean autoAckOldestChunkedMessageOnQueueFull; -// -// private int maxPendingChunkedMessage; -// -// private long expireTimeOfIncompleteChunkedMessageMillis; -// -// public PulsarConsumerProperties(String... topics) { -// this.topics = topics.clone(); -// this.topicsPattern = null; -// } -// -// public PulsarConsumerProperties(Pattern topicPattern) { -// this.topicsPattern = topicPattern; -// this.topics = null; -// } -// -// public void setTopics(String[] topics) { -// this.topics = topics; -// } -// -// public void setTopicsPattern(Pattern topicsPattern) { -// this.topicsPattern = topicsPattern; -// } -// -// public String getSubscriptionName() { -// return subscriptionName; -// } -// -// public void setSubscriptionName(String subscriptionName) { -// this.subscriptionName = subscriptionName; -// } -// -// public int getReceiveQueueSize() { -// return receiveQueueSize; -// } -// -// public void setReceiveQueueSize(int receiveQueueSize) { -// this.receiveQueueSize = receiveQueueSize; -// } -// -// public SubscriptionType getSubscriptionType() { -// return subscriptionType; -// } -// -// public void setSubscriptionType(SubscriptionType subscriptionType) { -// this.subscriptionType = subscriptionType; -// } -// -// public long getAcknowledgementsGroupTimeMicros() { -// return acknowledgementsGroupTimeMicros; -// } -// -// public void setAcknowledgementsGroupTimeMicros(long acknowledgementsGroupTimeMicros) { -// this.acknowledgementsGroupTimeMicros = acknowledgementsGroupTimeMicros; -// } -// -// public long getNegativeAckRedeliveryDelayMicros() { -// return negativeAckRedeliveryDelayMicros; -// } -// -// public void setNegativeAckRedeliveryDelayMicros(long negativeAckRedeliveryDelayMicros) { -// this.negativeAckRedeliveryDelayMicros = negativeAckRedeliveryDelayMicros; -// } -// -// public int getMaxTotalReceiverQueueSizeAcrossPartitions() { -// return maxTotalReceiverQueueSizeAcrossPartitions; -// } -// -// public void setMaxTotalReceiverQueueSizeAcrossPartitions(int maxTotalReceiverQueueSizeAcrossPartitions) { -// this.maxTotalReceiverQueueSizeAcrossPartitions = maxTotalReceiverQueueSizeAcrossPartitions; -// } -// -// public String getConsumerName() { -// return consumerName; -// } -// -// public void setConsumerName(String consumerName) { -// this.consumerName = consumerName; -// } -// -// public long getAckTimeoutMillis() { -// return ackTimeoutMillis; -// } -// -// public void setAckTimeoutMillis(long ackTimeoutMillis) { -// this.ackTimeoutMillis = ackTimeoutMillis; -// } -// -// public long getTickDurationMillis() { -// return tickDurationMillis; -// } -// -// public void setTickDurationMillis(long tickDurationMillis) { -// this.tickDurationMillis = tickDurationMillis; -// } -// -// public int getPriorityLevel() { -// return priorityLevel; -// } -// -// public void setPriorityLevel(int priorityLevel) { -// this.priorityLevel = priorityLevel; -// } -// -// public ConsumerCryptoFailureAction getCryptoFailureAction() { -// return cryptoFailureAction; -// } -// -// public void setCryptoFailureAction(ConsumerCryptoFailureAction cryptoFailureAction) { -// this.cryptoFailureAction = cryptoFailureAction; -// } -// -// public SortedMap getProperties() { -// return properties; -// } -// -// public void setProperties(SortedMap properties) { -// this.properties = properties; -// } -// -// public boolean isReadCompacted() { -// return readCompacted; -// } -// -// public void setReadCompacted(boolean readCompacted) { -// this.readCompacted = readCompacted; -// } -// -// public SubscriptionInitialPosition getSubscriptionInitialPosition() { -// return subscriptionInitialPosition; -// } -// -// public void setSubscriptionInitialPosition(SubscriptionInitialPosition subscriptionInitialPosition) { -// this.subscriptionInitialPosition = subscriptionInitialPosition; -// } -// -// public int getPatternAutoDiscoveryPeriod() { -// return patternAutoDiscoveryPeriod; -// } -// -// public void setPatternAutoDiscoveryPeriod(int patternAutoDiscoveryPeriod) { -// this.patternAutoDiscoveryPeriod = patternAutoDiscoveryPeriod; -// } -// -// public RegexSubscriptionMode getRegexSubscriptionMode() { -// return regexSubscriptionMode; -// } -// -// public void setRegexSubscriptionMode(RegexSubscriptionMode regexSubscriptionMode) { -// this.regexSubscriptionMode = regexSubscriptionMode; -// } -// -// public DeadLetterPolicy getDeadLetterPolicy() { -// return deadLetterPolicy; -// } -// -// public void setDeadLetterPolicy(DeadLetterPolicy deadLetterPolicy) { -// this.deadLetterPolicy = deadLetterPolicy; -// } -// -// public boolean isAutoUpdatePartitions() { -// return autoUpdatePartitions; -// } -// -// public void setAutoUpdatePartitions(boolean autoUpdatePartitions) { -// this.autoUpdatePartitions = autoUpdatePartitions; -// } -// -// public boolean isReplicateSubscriptionState() { -// return replicateSubscriptionState; -// } -// -// public void setReplicateSubscriptionState(boolean replicateSubscriptionState) { -// this.replicateSubscriptionState = replicateSubscriptionState; -// } -// -// public RedeliveryBackoff getNegativeAckRedeliveryBackoff() { -// return negativeAckRedeliveryBackoff; -// } -// -// public void setNegativeAckRedeliveryBackoff(RedeliveryBackoff negativeAckRedeliveryBackoff) { -// this.negativeAckRedeliveryBackoff = negativeAckRedeliveryBackoff; -// } -// -// public RedeliveryBackoff getAckTimeoutRedeliveryBackoff() { -// return ackTimeoutRedeliveryBackoff; -// } -// -// public void setAckTimeoutRedeliveryBackoff(RedeliveryBackoff ackTimeoutRedeliveryBackoff) { -// this.ackTimeoutRedeliveryBackoff = ackTimeoutRedeliveryBackoff; -// } -// -// public boolean isAutoAckOldestChunkedMessageOnQueueFull() { -// return autoAckOldestChunkedMessageOnQueueFull; -// } -// -// public void setAutoAckOldestChunkedMessageOnQueueFull(boolean autoAckOldestChunkedMessageOnQueueFull) { -// this.autoAckOldestChunkedMessageOnQueueFull = autoAckOldestChunkedMessageOnQueueFull; -// } -// -// public int getMaxPendingChunkedMessage() { -// return maxPendingChunkedMessage; -// } -// -// public void setMaxPendingChunkedMessage(int maxPendingChunkedMessage) { -// this.maxPendingChunkedMessage = maxPendingChunkedMessage; -// } -// -// public long getExpireTimeOfIncompleteChunkedMessageMillis() { -// return expireTimeOfIncompleteChunkedMessageMillis; -// } -// -// public void setExpireTimeOfIncompleteChunkedMessageMillis(long expireTimeOfIncompleteChunkedMessageMillis) { -// this.expireTimeOfIncompleteChunkedMessageMillis = expireTimeOfIncompleteChunkedMessageMillis; -// } -// -// public String[] getTopics() { -// return topics; -// } -// -// public Pattern getTopicsPattern() { -// return topicsPattern; -// } -// -// -//} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java index 78f0d7a6..d3882168 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java @@ -27,6 +27,8 @@ import org.springframework.core.task.AsyncListenableTaskExecutor; import org.springframework.util.Assert; /** + * Contains runtime properties for a listener container. + * * @author Soby Chacko */ public class PulsarContainerProperties { @@ -35,38 +37,41 @@ public class PulsarContainerProperties { private Duration consumerStartTimeout = DEFAULT_CONSUMER_START_TIMEOUT; + /** + * Enumeration for ack mode. + */ public enum AckMode { + /** + * Manual ack mode. + */ MANUAL; } - /** - * Topic names. - */ private String[] topics; - /** - * Topic pattern. - */ private Pattern topicsPattern; private String subscriptionName; private SubscriptionType subscriptionType; - private Schema schema; private SchemaType schemaType; private Object messageListener; + private AsyncListenableTaskExecutor consumerTaskExecutor; private int maxNumMessages = -1; + private int maxNumBytes = 10 * 1024 * 1024; + private int batchTimeout = 100; private boolean batchListener; + private boolean batchAsyncReceive; private boolean asyncReceive; @@ -84,7 +89,7 @@ public class PulsarContainerProperties { } public Object getMessageListener() { - return messageListener; + return this.messageListener; } public void setMessageListener(Object messageListener) { @@ -100,7 +105,7 @@ public class PulsarContainerProperties { } public SubscriptionType getSubscriptionType() { - return subscriptionType; + return this.subscriptionType; } public void setSubscriptionType(SubscriptionType subscriptionType) { @@ -108,7 +113,7 @@ public class PulsarContainerProperties { } public int getMaxNumMessages() { - return maxNumMessages; + return this.maxNumMessages; } public void setMaxNumMessages(int maxNumMessages) { @@ -116,7 +121,7 @@ public class PulsarContainerProperties { } public int getMaxNumBytes() { - return maxNumBytes; + return this.maxNumBytes; } public void setMaxNumBytes(int maxNumBytes) { @@ -124,7 +129,7 @@ public class PulsarContainerProperties { } public int getBatchTimeout() { - return batchTimeout; + return this.batchTimeout; } public void setBatchTimeout(int batchTimeout) { @@ -132,7 +137,7 @@ public class PulsarContainerProperties { } public boolean isBatchListener() { - return batchListener; + return this.batchListener; } public void setBatchListener(boolean batchListener) { @@ -140,7 +145,7 @@ public class PulsarContainerProperties { } public boolean isBatchAsyncReceive() { - return batchAsyncReceive; + return this.batchAsyncReceive; } public void setBatchAsyncReceive(boolean batchAsyncReceive) { @@ -148,7 +153,7 @@ public class PulsarContainerProperties { } public boolean isAsyncReceive() { - return asyncReceive; + return this.asyncReceive; } public void setAsyncReceive(boolean asyncReceive) { @@ -156,7 +161,7 @@ public class PulsarContainerProperties { } public AckMode getAckMode() { - return ackMode; + return this.ackMode; } public void setAckMode(AckMode ackMode) { @@ -178,7 +183,7 @@ public class PulsarContainerProperties { } public Schema getSchema() { - return schema; + return this.schema; } public void setSchema(Schema schema) { @@ -186,7 +191,7 @@ public class PulsarContainerProperties { } public String[] getTopics() { - return topics; + return this.topics; } public void setTopics(String[] topics) { @@ -194,7 +199,7 @@ public class PulsarContainerProperties { } public Pattern getTopicsPattern() { - return topicsPattern; + return this.topicsPattern; } public void setTopicsPattern(Pattern topicsPattern) { @@ -202,7 +207,7 @@ public class PulsarContainerProperties { } public String getSubscriptionName() { - return subscriptionName; + return this.subscriptionName; } public void setSubscriptionName(String subscriptionName) { @@ -210,7 +215,7 @@ public class PulsarContainerProperties { } public SchemaType getSchemaType() { - return schemaType; + return this.schemaType; } public void setSchemaType(SchemaType schemaType) { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarListenerContainerRegistry.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarListenerContainerRegistry.java index 4b5e8074..cd29b113 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarListenerContainerRegistry.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarListenerContainerRegistry.java @@ -20,9 +20,10 @@ import java.util.Collection; import java.util.Set; import org.springframework.lang.Nullable; -import org.springframework.pulsar.listener.PulsarMessageListenerContainer; /** + * A registry for listener containers. + * * @author Soby Chacko */ public interface PulsarListenerContainerRegistry { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageListenerContainer.java index 70bcef65..656fbf53 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageListenerContainer.java @@ -20,6 +20,9 @@ import org.springframework.beans.factory.DisposableBean; import org.springframework.context.SmartLifecycle; /** + * Internal abstraction used by the framework representing a message + * listener container. Not meant to be implemented externally. + * * @author Soby Chacko */ public interface PulsarMessageListenerContainer extends SmartLifecycle, DisposableBean { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/DelegatingInvocableHandler.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/DelegatingInvocableHandler.java index b8620e6c..753a11c7 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/DelegatingInvocableHandler.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/DelegatingInvocableHandler.java @@ -41,10 +41,14 @@ import org.springframework.messaging.handler.annotation.Header; import org.springframework.messaging.handler.annotation.support.PayloadMethodArgumentResolver; import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; import org.springframework.pulsar.PulsarException; -import org.springframework.pulsar.listener.adapter.InvocationResult; import org.springframework.validation.Validator; /** + * Delegates to an {@link InvocableHandlerMethod} based on the message payload type. + * Matches a single, non-annotated parameter or one that is annotated with + * {@link org.springframework.messaging.handler.annotation.Payload}. Matches must be + * unambiguous. + * * @author Soby Chacko */ public class DelegatingInvocableHandler { @@ -75,11 +79,10 @@ public class DelegatingInvocableHandler { private final PayloadValidator validator; public DelegatingInvocableHandler(List handlers, - @Nullable InvocableHandlerMethod defaultHandler, Object bean, - @Nullable BeanExpressionResolver beanExpressionResolver, - @Nullable BeanExpressionContext beanExpressionContext, - @Nullable BeanFactory beanFactory, @Nullable Validator validator) { - + @Nullable InvocableHandlerMethod defaultHandler, Object bean, + @Nullable BeanExpressionResolver beanExpressionResolver, + @Nullable BeanExpressionContext beanExpressionContext, + @Nullable BeanFactory beanFactory, @Nullable Validator validator) { this.handlers = new ArrayList<>(); for (InvocableHandlerMethod handler : handlers) { this.handlers.add(wrapIfNecessary(handler)); @@ -210,7 +213,7 @@ public class DelegatingInvocableHandler { } private MethodParameter findCandidate(Class payloadClass, Method method, - Annotation[][] parameterAnnotations) { + Annotation[][] parameterAnnotations) { MethodParameter foundCandidate = null; for (int i = 0; i < parameterAnnotations.length; i++) { MethodParameter methodParameter = new MethodParameter(method, i); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/HandlerAdapter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/HandlerAdapter.java index 78649f3d..e5cb4657 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/HandlerAdapter.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/HandlerAdapter.java @@ -20,6 +20,10 @@ import org.springframework.messaging.Message; import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; /** + * A wrapper for either an {@link InvocableHandlerMethod} or + * {@link DelegatingInvocableHandler}. All methods delegate to the + * underlying handler. + * * @author Soby Chacko */ public class HandlerAdapter { @@ -81,7 +85,7 @@ public class HandlerAdapter { } public InvocableHandlerMethod getInvokerHandlerMethod() { - return invokerHandlerMethod; + return this.invokerHandlerMethod; } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/InvocationResult.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/InvocationResult.java index 9b6d3eab..9b37b771 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/InvocationResult.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/InvocationResult.java @@ -20,6 +20,8 @@ import org.springframework.expression.Expression; import org.springframework.lang.Nullable; /** + * The result of a method invocation. + * * @author Soby Chacko */ public final class InvocationResult { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarBatchMessagingMessageListenerAdapter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarBatchMessagingMessageListenerAdapter.java index a981f132..413b588e 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarBatchMessagingMessageListenerAdapter.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarBatchMessagingMessageListenerAdapter.java @@ -25,16 +25,23 @@ import org.apache.pulsar.client.api.Messages; import org.springframework.messaging.Message; import org.springframework.messaging.support.MessageBuilder; -import org.springframework.pulsar.support.converter.PulsarBatchMessageConverter; import org.springframework.pulsar.listener.PulsarBatchMessageListener; +import org.springframework.pulsar.support.converter.PulsarBatchMessageConverter; import org.springframework.pulsar.support.converter.PulsarBatchMessagingMessageConverter; import org.springframework.pulsar.support.converter.PulsarRecordMessageConverter; import org.springframework.util.Assert; /** + * A {@link org.apache.pulsar.client.api.MessageListener MessageListener} + * adapter that invokes a configurable {@link HandlerAdapter}; used when the factory is + * configured for the listener to receive batches of messages. + * + * @param payload type. + * * @author Soby Chacko */ -public class PulsarBatchMessagingMessageListenerAdapter extends PulsarMessagingMessageListenerAdapter +@SuppressWarnings("serial") +public class PulsarBatchMessagingMessageListenerAdapter extends PulsarMessagingMessageListenerAdapter implements PulsarBatchMessageListener { private PulsarBatchMessageConverter batchMessageConverter = new PulsarBatchMessagingMessageConverter(); @@ -78,7 +85,7 @@ public class PulsarBatchMessagingMessageListenerAdapter extends PulsarMessag } protected void invoke(Object records, Consumer consumer, - final Message messageArg) { + final Message messageArg) { Message message = messageArg; try { @@ -92,7 +99,6 @@ public class PulsarBatchMessagingMessageListenerAdapter extends PulsarMessag } } - protected Message toMessagingMessage(Messages msg, Consumer consumer) { return getBatchMessageConverter().toMessage(msg, consumer, getType()); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarMessagingMessageListenerAdapter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarMessagingMessageListenerAdapter.java index f8176f3e..5c86da80 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarMessagingMessageListenerAdapter.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarMessagingMessageListenerAdapter.java @@ -43,6 +43,12 @@ import org.springframework.pulsar.support.converter.PulsarRecordMessageConverter import org.springframework.util.Assert; /** + * An abstract {@link org.apache.pulsar.client.api.MessageListener} adapter + * providing the necessary infrastructure to extract the payload from a + * Pulsar message. + * + * @param payload type. + * * @author Soby Chacko */ public abstract class PulsarMessagingMessageListenerAdapter { @@ -134,7 +140,7 @@ public abstract class PulsarMessagingMessageListenerAdapter { } protected final Object invokeHandler(Object data, org.springframework.messaging.Message message, - Consumer consumer) { + Consumer consumer) { try { return this.handlerMethod.invoke(message, data, consumer); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarRecordMessagingMessageListenerAdapter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarRecordMessagingMessageListenerAdapter.java index 8f74eaa9..da113b0b 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarRecordMessagingMessageListenerAdapter.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarRecordMessagingMessageListenerAdapter.java @@ -23,8 +23,15 @@ import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.MessageListener; /** + * A {@link MessageListener MessageListener} + * adapter that invokes a configurable {@link HandlerAdapter}; used when the factory is + * configured for the listener to receive individual messages. + * + * @param payload type. + * * @author Soby Chacko */ +@SuppressWarnings("serial") public class PulsarRecordMessagingMessageListenerAdapter extends PulsarMessagingMessageListenerAdapter implements MessageListener { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/support/EndpointHandlerMethod.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/EndpointHandlerMethod.java index fc4b621e..b4b90e58 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/support/EndpointHandlerMethod.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/support/EndpointHandlerMethod.java @@ -28,6 +28,8 @@ import org.springframework.util.Assert; import org.springframework.util.ReflectionUtils; /** + * Handler method for endpoints. + * * @author Soby Chacko */ public class EndpointHandlerMethod { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/support/JavaUtils.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/JavaUtils.java index c73e827b..65fbc0d5 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/support/JavaUtils.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/support/JavaUtils.java @@ -26,6 +26,9 @@ import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; /** + * Chained utility methods to simplify some Java repetitive code. Obtain a reference to + * the singleton {@link #INSTANCE} and then chain calls to the utility methods. + * * @author Soby Chacko */ public final class JavaUtils { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/support/MessageConverter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/MessageConverter.java index d04340d8..7cbf7a27 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/support/MessageConverter.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/support/MessageConverter.java @@ -17,6 +17,8 @@ package org.springframework.pulsar.support; /** + * Pulsar specific message converter. + * * @author Soby Chacko */ public interface MessageConverter { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarBatchMessageConverter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarBatchMessageConverter.java index 682ddc46..4ef43400 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarBatchMessageConverter.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarBatchMessageConverter.java @@ -21,12 +21,14 @@ import java.lang.reflect.Type; import org.apache.pulsar.client.api.Consumer; import org.apache.pulsar.client.api.Messages; -import org.springframework.lang.NonNull; -import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.pulsar.support.MessageConverter; /** + * Pulsar batch message converter strategy. + * + * @param message type. + * * @author Soby Chacko */ public interface PulsarBatchMessageConverter extends MessageConverter { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarBatchMessagingMessageConverter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarBatchMessagingMessageConverter.java index 268c0bf6..24b492fa 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarBatchMessagingMessageConverter.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarBatchMessagingMessageConverter.java @@ -28,17 +28,18 @@ import org.apache.pulsar.client.api.Messages; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.support.MessageBuilder; -import org.springframework.pulsar.support.converter.PulsarBatchMessageConverter; -import org.springframework.pulsar.support.converter.PulsarRecordMessageConverter; /** + * Batch records message converter. + * + * @param message type. + * * @author Soby Chacko */ public class PulsarBatchMessagingMessageConverter implements PulsarBatchMessageConverter { private final PulsarRecordMessageConverter recordConverter; - public PulsarBatchMessagingMessageConverter() { this(null); } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarMessagingMessageConverter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarMessagingMessageConverter.java index f3cac499..c04dafe8 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarMessagingMessageConverter.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarMessagingMessageConverter.java @@ -29,6 +29,12 @@ import org.springframework.messaging.support.GenericMessage; import org.springframework.messaging.support.MessageBuilder; /** + * + * A Messaging {@link org.springframework.pulsar.support.MessageConverter} implementation for a message listener that + * receives individual messages. + * + * @param message type + * * @author Soby Chacko */ public class PulsarMessagingMessageConverter implements PulsarRecordMessageConverter { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarRecordMessageConverter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarRecordMessageConverter.java index dcdbf3e5..93a05f8f 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarRecordMessageConverter.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarRecordMessageConverter.java @@ -25,6 +25,10 @@ import org.springframework.messaging.Message; import org.springframework.pulsar.support.MessageConverter; /** + * Pulsar specific record converter strategy. + * + * @param message type + * * @author Soby Chacko */ public interface PulsarRecordMessageConverter extends MessageConverter { @@ -32,7 +36,7 @@ public interface PulsarRecordMessageConverter extends MessageConverter { @NonNull Message toMessage(org.apache.pulsar.client.api.Message record, Consumer consumer, - Type payloadType); + Type payloadType); T fromMessage(Message message, String defaultTopic); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/AbstractContainerBaseTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/AbstractContainerBaseTests.java index 748d63b6..f4816432 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/AbstractContainerBaseTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/AbstractContainerBaseTests.java @@ -1,3 +1,19 @@ +/* + * 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.core; import org.testcontainers.containers.PulsarContainer; diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/FailoverConsumerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/FailoverConsumerTests.java index 2badd573..b4b2180b 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/FailoverConsumerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/FailoverConsumerTests.java @@ -16,6 +16,8 @@ package org.springframework.pulsar.core; +import static org.assertj.core.api.Assertions.assertThat; + import java.io.Serial; import java.util.HashMap; import java.util.HashSet; @@ -37,8 +39,6 @@ import org.junit.jupiter.api.Test; import org.springframework.pulsar.listener.DefaultPulsarMessageListenerContainer; import org.springframework.pulsar.listener.PulsarContainerProperties; -import static org.assertj.core.api.AssertionsForClassTypes.assertThat; - /** * @author Soby Chacko */ diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java index 5a2aaa2a..d3ab6d6e 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java @@ -16,6 +16,8 @@ package org.springframework.pulsar.core; +import static org.assertj.core.api.Assertions.assertThat; + import java.util.HashMap; import java.util.Map; import java.util.concurrent.CompletableFuture; @@ -30,8 +32,6 @@ import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.Schema; import org.junit.jupiter.api.Test; -import static org.assertj.core.api.AssertionsForClassTypes.assertThat; - /** * @author Soby Chacko */ @@ -62,11 +62,10 @@ class PulsarTemplateTests extends AbstractContainerBaseTests { final DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(client, config); final PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); final CompletableFuture future = pulsarTemplate.sendAsync("hello john doe"); - future.thenAccept(m -> System.out.println("Got " + m)); + future.thenAccept(m -> { }); try { Thread.sleep(2000); - final MessageId messageId = future.get(); - System.out.println(); + future.get(); } catch (InterruptedException | ExecutionException e) { e.printStackTrace();