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

Build cleanup (#19)

Browse Source 
* Build cleanup

* Addressing PR review comments
This commit is contained in:
Soby Chacko
2022-07-08 18:27:30 -04:00
committed by GitHub
parent 1777de2aea
commit 100e3f3763
66 changed files with 544 additions and 945 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
README.adoc
View File

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

132
build.gradle
View File

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

52
publish-maven.gradle Normal file
View File

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

2
spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java
View File

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

6
spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java
View File

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

16
spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java
View File

@@ -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.
* <p>
* 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);

18
spring-pulsar-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/AbstractContainerBaseTests.java
View File

@@ -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");

4
spring-pulsar-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarListenerTests.java
View File

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

26
spring-pulsar-sample-apps/src/main/java/app1/PulsarBootApp.java
View File

@@ -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 + '\'' +
'}';
}
}

19
spring-pulsar-sample-apps/src/main/java/app2/ProducerApp.java
View File

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

66
spring-pulsar-sample-apps/src/main/java/app3/PulsarAppTry.java
View File

@@ -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<String> pulsarProducerFactory(PulsarClient pulsarClient) {
// Map<String, Object> 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<String> pulsarTemplate(PulsarProducerFactory<String> pulsarProducerFactory) {
// return new PulsarTemplate<>(pulsarProducerFactory);
// }
//
// @Bean
// public PulsarConsumerFactory<?> pulsarConsumerFactory(PulsarClient pulsarClient) {
//
// Map<String, Object> config = new HashMap<>();
//// final HashSet<String> 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<Object> 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 {
//
// }
}

24
spring-pulsar-sample-apps/src/main/java/app4/FailoverConsumerApp.java
View File

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

18
spring-pulsar-sample-apps/src/main/java/app5/FailoverConsumer.java
View File

@@ -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);
//...
}

71
spring-pulsar-sample-apps/src/main/java/app6/Sender.java
View File

@@ -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<byte[]> template;
public Sender(PulsarTemplate<byte[]> template) {
this.template = template;
}
public void send() throws PulsarClientException {
this.template.send("foobar-fuzzy".getBytes(StandardCharsets.UTF_8));
}
}
@Configuration
class Config {
@Bean
public PulsarProducerFactory<byte[]> pulsarProducerFactory(PulsarClient pulsarClient) {
Map<String, Object> config = new HashMap<>();
config.put("topicName", "foo-1");
return new DefaultPulsarProducerFactory<>(pulsarClient, config);
}
@Bean
public PulsarTemplate<byte[]> pulsarTemplate(PulsarProducerFactory<byte[]> pulsarProducerFactory) {
return new PulsarTemplate<>(pulsarProducerFactory);
}
@Bean
public Sender sender(PulsarTemplate<byte[]> template) {
return new Sender(template);
}
@Bean
public PulsarClientFactoryBean pulsarClientFactoryBean(PulsarClientConfiguration pulsarClientConfiguration) {
return new PulsarClientFactoryBean(pulsarClientConfiguration);
}
@Bean
public PulsarClientConfiguration pulsarClientConfiguration() {
return new PulsarClientConfiguration();
}
}

83
spring-pulsar-sample-apps/src/main/java/app6/SenderString.java
View File

@@ -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<String> template;
public SenderString(PulsarTemplate<String> template) {
this.template = template;
}
public void send() throws PulsarClientException {
final CompletableFuture<MessageId> 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<String> pulsarProducerFactory(PulsarClient pulsarClient) {
Map<String, Object> 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<String> pulsarTemplate(PulsarProducerFactory<String> pulsarProducerFactory) {
return new PulsarTemplate<>(pulsarProducerFactory);
}
@Bean
public SenderString senderString(PulsarTemplate<String> template) {
return new SenderString(template);
}
}

3
spring-pulsar/src/main/java/org/springframework/pulsar/PulsarException.java
View File

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

5
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarBootstrapConfiguration.java
View File

@@ -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}.
*
* <p>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
}
}
}
}

1
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java
View File

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

11
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java
View File

@@ -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<K, V> 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<K, V> 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<K, V> 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<K, V> 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);

2
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerConfigurer.java
View File

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

6
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListeners.java
View File

@@ -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();
}
}

10
spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java
View File

@@ -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 <C> the {@link AbstractPulsarMessageListenerContainer} implementation type.
* @param <T> Message payload type.
*
* @author Soby Chacko
*/
public abstract class AbstractPulsarListenerContainerFactory<C extends AbstractPulsarMessageListenerContainer<T>, T>

16
spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerEndpoint.java
View File

@@ -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 <K> Message payload type.
*
* @author Soby Chacko
*/
public abstract class AbstractPulsarListenerEndpoint<K> implements PulsarListenerEndpoint, BeanFactoryAware, InitializingBean {
@@ -152,14 +156,14 @@ public abstract class AbstractPulsarListenerEndpoint<K> 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<K> adapter = createMessageListener(container, messageConverter);
Object messageListener = adapter;
@@ -170,7 +174,7 @@ public abstract class AbstractPulsarListenerEndpoint<K> implements PulsarListene
}
protected abstract PulsarMessagingMessageListenerAdapter<K> 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<K> implements PulsarListene
public SubscriptionType getSubscriptionType() {
return subscriptionType;
return this.subscriptionType;
}
public void setSubscriptionType(SubscriptionType subscriptionType) {
@@ -200,7 +204,7 @@ public abstract class AbstractPulsarListenerEndpoint<K> implements PulsarListene
}
public SchemaType getSchemaType() {
return schemaType;
return this.schemaType;
}
public void setSchemaType(SchemaType schemaType) {

8
spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java
View File

@@ -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 <V> Message payload type
*
* @author Soby Chacko
*/
public class MethodPulsarListenerEndpoint<V> extends AbstractPulsarListenerEndpoint<V> {
@@ -91,7 +96,7 @@ public class MethodPulsarListenerEndpoint<V> extends AbstractPulsarListenerEndpo
@Override
protected PulsarMessagingMessageListenerAdapter<V> createMessageListener(PulsarMessageListenerContainer container,
@Nullable MessageConverter messageConverter) {
@Nullable MessageConverter messageConverter) {
Assert.state(this.messageHandlerMethodFactory != null,
"Could not create message listener - MessageHandlerMethodFactory not set");
PulsarMessagingMessageListenerAdapter<V> messageListener = createMessageListenerInstance(messageConverter);
@@ -226,5 +231,4 @@ public class MethodPulsarListenerEndpoint<V> extends AbstractPulsarListenerEndpo
this.messagingConverter = messagingConverter;
}
}

4
spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarClientConfiguration.java
View File

@@ -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<String, Object> getConfigs() {
return configs;
return this.configs;
}
}

6
spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarClientFactoryBean.java
View File

@@ -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<PulsarClient> {
@@ -48,7 +48,7 @@ public class PulsarClientFactoryBean extends AbstractFactoryBean<PulsarClient> {
@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();
}
}

2
spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerConfigUtils.java
View File

@@ -17,6 +17,8 @@
package org.springframework.pulsar.config;
/**
* Configuration constants for internal sharing across subpackages.
*
* @author Soby Chacko
*/
public abstract class PulsarListenerConfigUtils {

4
spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerContainerFactory.java
View File

@@ -19,6 +19,10 @@ package org.springframework.pulsar.config;
import org.springframework.pulsar.listener.PulsarMessageListenerContainer;
/**
* Factory for Pulsar message listener containers.
*
* @param <C> message listener container type.
*
* @author Soby Chacko
*/
public interface PulsarListenerContainerFactory<C extends PulsarMessageListenerContainer> {

7
spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerContainerFactoryImpl.java
View File

@@ -26,6 +26,11 @@ import org.springframework.pulsar.listener.PulsarContainerProperties;
import org.springframework.util.StringUtils;
/**
* Concrete implementation for {@link PulsarListenerContainerFactory}.
*
* @param <C> container implementation type.
* @param <T> message type in the listener.
*
* @author Soby Chacko
*/
public class PulsarListenerContainerFactoryImpl<C, T> extends AbstractPulsarListenerContainerFactory<DefaultPulsarMessageListenerContainer<T>, T> {
@@ -61,7 +66,7 @@ public class PulsarListenerContainerFactoryImpl<C, T> extends AbstractPulsarList
@Override
protected void initializeContainer(DefaultPulsarMessageListenerContainer<T> instance,
PulsarListenerEndpoint endpoint) {
PulsarListenerEndpoint endpoint) {
super.initializeContainer(instance, endpoint);
}

6
spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpoint.java
View File

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

4
spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointAdapter.java
View File

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

11
spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointRegistrar.java
View File

@@ -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<PulsarListenerEndpointDescriptor> endpointDescriptors = new ArrayList<>();
private PulsarListenerEndpointRegistry endpointRegistry;
private PulsarListenerEndpointRegistry endpointRegistry;
private List<HandlerMethodArgumentResolver> 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;

16
spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointRegistry.java
View File

@@ -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.
*
* <p>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;

10
spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java
View File

@@ -31,6 +31,10 @@ import org.apache.pulsar.client.api.Schema;
import org.springframework.util.CollectionUtils;
/**
* Default implementation for {@link PulsarConsumerFactory}.
*
* @param <T> underlying payload type for the consumer.
*
* @author Soby Chacko
*/
public class DefaultPulsarConsumerFactory<T> implements PulsarConsumerFactory<T> {
@@ -60,7 +64,7 @@ public class DefaultPulsarConsumerFactory<T> implements PulsarConsumerFactory<T>
consumerBuilder.loadConf(properties);
}
Consumer<T> consumer = consumerBuilder.subscribe();
consumers.add(consumer);
this.consumers.add(consumer);
return consumer;
}
@@ -77,12 +81,12 @@ public class DefaultPulsarConsumerFactory<T> implements PulsarConsumerFactory<T>
consumerBuilder.batchReceivePolicy(batchReceivePolicy);
Consumer<T> consumer = consumerBuilder.subscribe();
consumers.add(consumer);
this.consumers.add(consumer);
return consumer;
}
public Map<String, Object> getConsumerConfig() {
return consumerConfig;
return this.consumerConfig;
}
}

10
spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java
View File

@@ -32,6 +32,10 @@ import org.springframework.core.log.LogAccessor;
import org.springframework.util.CollectionUtils;
/**
* Default implementation for {@link PulsarProducerFactory}.
*
* @param <T> producer type.
*
* @author Soby Chacko
*/
public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>, DisposableBean {
@@ -60,7 +64,7 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
producerBuilder.loadConf(this.producerConfig);
}
this.producer = producerBuilder.create();
return producer;
return this.producer;
}
@Override
@@ -73,12 +77,12 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
}
producerBuilder.messageRouter(messageRouter);
this.producer = producerBuilder.create();
return producer;
return this.producer;
}
@Override
public Map<String, Object> getProducerConfig() {
return producerConfig;
return this.producerConfig;
}
@Override

245
spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarClientProperties.java
View File

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

4
spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarConsumerFactory.java
View File

@@ -24,6 +24,10 @@ import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
/**
* Pulsar consumer factory interface.
*
* @param <T> payload type for the consumer.
*
* @author Soby Chacko
*/
public interface PulsarConsumerFactory<T> {

5
spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java
View File

@@ -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 <T> producer payload type.
*
* @author Soby Chacko
*/
public interface PulsarProducerFactory<T> {

34
spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java
View File

@@ -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 <T> message type.
*
* @author Soby Chacko
*/
public class PulsarTemplate<T> {
@@ -47,10 +51,10 @@ public class PulsarTemplate<T> {
public MessageId send(T message) throws PulsarClientException {
final Schema<T> schema = this.schema != null ? this.schema : SchemaUtils.getSchema(message);
final SchemaTopic schemaTopic = getSchemaTopic(schema, this.pulsarProducerFactory, null);
Producer<T> producer = producerCache.get(schemaTopic);
Producer<T> 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<T> {
public CompletableFuture<MessageId> sendAsync(T message) throws PulsarClientException {
final Schema<T> schema = this.schema != null ? this.schema : SchemaUtils.getSchema(message);
final SchemaTopic schemaTopic = getSchemaTopic(schema, this.pulsarProducerFactory, null);
Producer<T> producer = producerCache.get(schemaTopic);
Producer<T> 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<T> {
public CompletableFuture<MessageId> sendAsync(T message, MessageRouter messageRouter) throws PulsarClientException {
final Schema<T> schema = this.schema != null ? this.schema : SchemaUtils.getSchema(message);
final SchemaTopic schemaTopic = getSchemaTopic(schema, this.pulsarProducerFactory, messageRouter);
Producer<T> producer = producerCache.get(schemaTopic);
Producer<T> 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<T> {
}
public Schema<T> getSchema() {
return schema;
return this.schema;
}
public void setSchema(Schema<T> schema) {
@@ -100,7 +104,7 @@ public class PulsarTemplate<T> {
final String topicName;
final MessageRouter messageRouter;
public SchemaTopic(Schema<T> schema, String topicName, MessageRouter messageRouter) {
SchemaTopic(Schema<T> schema, String topicName, MessageRouter messageRouter) {
this.schema = schema;
this.topicName = topicName;
this.messageRouter = messageRouter;
@@ -108,12 +112,16 @@ public class PulsarTemplate<T> {
@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<T> {
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);
}
}
}

10
spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaUtils.java
View File

@@ -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 <T> Schema<T> getSchema(T message) {

2
spring-pulsar/src/main/java/org/springframework/pulsar/event/ConsumerFailedToStartEvent.java
View File

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

2
spring-pulsar/src/main/java/org/springframework/pulsar/event/ConsumerStartedEvent.java
View File

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

2
spring-pulsar/src/main/java/org/springframework/pulsar/event/ConsumerStartingEvent.java
View File

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

2
spring-pulsar/src/main/java/org/springframework/pulsar/event/PulsarEvent.java
View File

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

13
spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java
View File

@@ -29,6 +29,10 @@ import org.springframework.lang.Nullable;
import org.springframework.pulsar.core.PulsarConsumerFactory;
/**
* Base implementation for the {@link PulsarMessageListenerContainer}.
*
* @param <T> message type.
*
* @author Soby Chacko
*/
public abstract class AbstractPulsarMessageListenerContainer<T>
@@ -49,9 +53,10 @@ public abstract class AbstractPulsarMessageListenerContainer<T>
private int phase;
@SuppressWarnings("unchecked")
protected AbstractPulsarMessageListenerContainer(PulsarConsumerFactory<? super T> pulsarConsumerFactory,PulsarContainerProperties pulsarContainerProperties) {
protected AbstractPulsarMessageListenerContainer(PulsarConsumerFactory<? super T> pulsarConsumerFactory,
PulsarContainerProperties pulsarContainerProperties) {
this.pulsarContainerProperties = pulsarContainerProperties;
this.pulsarConsumerFactory = (PulsarConsumerFactory<T>)pulsarConsumerFactory;
this.pulsarConsumerFactory = (PulsarConsumerFactory<T>) pulsarConsumerFactory;
}
@@ -99,11 +104,11 @@ public abstract class AbstractPulsarMessageListenerContainer<T>
}
public PulsarContainerProperties getPulsarContainerProperties() {
return pulsarContainerProperties;
return this.pulsarContainerProperties;
}
public PulsarConsumerFactory<? super T> getPulsarConsumerFactory() {
return pulsarConsumerFactory;
return this.pulsarConsumerFactory;
}
@Override

10
spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java
View File

@@ -45,6 +45,10 @@ import org.springframework.util.StringUtils;
import org.springframework.util.concurrent.ListenableFuture;
/**
* Default implementation for {@link PulsarMessageListenerContainer}.
*
* @param <T> message type.
*
* @author Soby Chacko
*/
public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMessageListenerContainer<T> {
@@ -243,11 +247,11 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
Messages<T> 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<T> extends AbstractPulsarMess
else {
for (Message<T> 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);
}

309
spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarConsumerProperties.java
View File

@@ -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<String, String> 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<String, String> getProperties() {
// return properties;
// }
//
// public void setProperties(SortedMap<String, String> 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;
// }
//
//
//}

47
spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java
View File

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

3
spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarListenerContainerRegistry.java
View File

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

3
spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageListenerContainer.java
View File

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

17
spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/DelegatingInvocableHandler.java
View File

@@ -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<InvocableHandlerMethod> 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<? extends Object> payloadClass, Method method,
Annotation[][] parameterAnnotations) {
Annotation[][] parameterAnnotations) {
MethodParameter foundCandidate = null;
for (int i = 0; i < parameterAnnotations.length; i++) {
MethodParameter methodParameter = new MethodParameter(method, i);

6
spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/HandlerAdapter.java
View File

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

2
spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/InvocationResult.java
View File

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

14
spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarBatchMessagingMessageListenerAdapter.java
View File

@@ -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 <V> payload type.
*
* @author Soby Chacko
*/
public class PulsarBatchMessagingMessageListenerAdapter <V> extends PulsarMessagingMessageListenerAdapter<V>
@SuppressWarnings("serial")
public class PulsarBatchMessagingMessageListenerAdapter<V> extends PulsarMessagingMessageListenerAdapter<V>
implements PulsarBatchMessageListener<V> {
private PulsarBatchMessageConverter<V> batchMessageConverter = new PulsarBatchMessagingMessageConverter<V>();
@@ -78,7 +85,7 @@ public class PulsarBatchMessagingMessageListenerAdapter <V> extends PulsarMessag
}
protected void invoke(Object records, Consumer<V> consumer,
final Message<?> messageArg) {
final Message<?> messageArg) {
Message<?> message = messageArg;
try {
@@ -92,7 +99,6 @@ public class PulsarBatchMessagingMessageListenerAdapter <V> extends PulsarMessag
}
}
protected Message<?> toMessagingMessage(Messages<V> msg, Consumer<V> consumer) {
return getBatchMessageConverter().toMessage(msg, consumer, getType());

8
spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarMessagingMessageListenerAdapter.java
View File

@@ -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 <V> payload type.
*
* @author Soby Chacko
*/
public abstract class PulsarMessagingMessageListenerAdapter<V> {
@@ -134,7 +140,7 @@ public abstract class PulsarMessagingMessageListenerAdapter<V> {
}
protected final Object invokeHandler(Object data, org.springframework.messaging.Message<?> message,
Consumer<V> consumer) {
Consumer<V> consumer) {
try {
return this.handlerMethod.invoke(message, data, consumer);

7
spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarRecordMessagingMessageListenerAdapter.java
View File

@@ -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 <V> payload type.
*
* @author Soby Chacko
*/
@SuppressWarnings("serial")
public class PulsarRecordMessagingMessageListenerAdapter<V> extends PulsarMessagingMessageListenerAdapter<V>
implements MessageListener<V> {

2
spring-pulsar/src/main/java/org/springframework/pulsar/support/EndpointHandlerMethod.java
View File

@@ -28,6 +28,8 @@ import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
/**
* Handler method for endpoints.
*
* @author Soby Chacko
*/
public class EndpointHandlerMethod {

3
spring-pulsar/src/main/java/org/springframework/pulsar/support/JavaUtils.java
View File

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

2
spring-pulsar/src/main/java/org/springframework/pulsar/support/MessageConverter.java
View File

@@ -17,6 +17,8 @@
package org.springframework.pulsar.support;
/**
* Pulsar specific message converter.
*
* @author Soby Chacko
*/
public interface MessageConverter {

6
spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarBatchMessageConverter.java
View File

@@ -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 <T> message type.
*
* @author Soby Chacko
*/
public interface PulsarBatchMessageConverter<T> extends MessageConverter {

7
spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarBatchMessagingMessageConverter.java
View File

@@ -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 <T> message type.
*
* @author Soby Chacko
*/
public class PulsarBatchMessagingMessageConverter<T> implements PulsarBatchMessageConverter<T> {
private final PulsarRecordMessageConverter<T> recordConverter;
public PulsarBatchMessagingMessageConverter() {
this(null);
}

6
spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarMessagingMessageConverter.java
View File

@@ -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 <V> message type
*
* @author Soby Chacko
*/
public class PulsarMessagingMessageConverter<V> implements PulsarRecordMessageConverter<V> {

6
spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarRecordMessageConverter.java
View File

@@ -25,6 +25,10 @@ import org.springframework.messaging.Message;
import org.springframework.pulsar.support.MessageConverter;
/**
* Pulsar specific record converter strategy.
*
* @param <T> message type
*
* @author Soby Chacko
*/
public interface PulsarRecordMessageConverter<T> extends MessageConverter {
@@ -32,7 +36,7 @@ public interface PulsarRecordMessageConverter<T> extends MessageConverter {
@NonNull
Message<?> toMessage(org.apache.pulsar.client.api.Message<T> record, Consumer<T> consumer,
Type payloadType);
Type payloadType);
T fromMessage(Message<?> message, String defaultTopic);

16
spring-pulsar/src/test/java/org/springframework/pulsar/core/AbstractContainerBaseTests.java
View File

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

4
spring-pulsar/src/test/java/org/springframework/pulsar/core/FailoverConsumerTests.java
View File

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

9
spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java
View File

@@ -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<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(client, config);
final PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
final CompletableFuture<MessageId> 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();