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

Refactor template/producer to not use default topic name (#25)

Browse Source
This commit is contained in:
Chris Bono
2022-07-11 19:51:21 -05:00
committed by GitHub
parent ecd0a30c83
commit 03425c550a
8 changed files with 178 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -51,8 +51,7 @@ class PulsarListenerTests extends AbstractContainerBaseTests {
try (ConfigurableApplicationContext context = app.run("--spring.pulsar.client.serviceUrl=" + AbstractContainerBaseTests.getPulsarBrokerUrl())) {
@SuppressWarnings("unchecked")
final PulsarTemplate<String> pulsarTemplate = context.getBean(PulsarTemplate.class);
pulsarTemplate.setDefaultTopicName("hello-pulsar-exclusive");
pulsarTemplate.send("John Doe");
pulsarTemplate.send("hello-pulsar-exclusive", "John Doe");
final boolean await = latch1.await(20, TimeUnit.SECONDS);
assertThat(await).isTrue();
}
@@ -66,9 +65,8 @@ class PulsarListenerTests extends AbstractContainerBaseTests {
try (ConfigurableApplicationContext context = app.run("--spring.pulsar.client.serviceUrl=" + AbstractContainerBaseTests.getPulsarBrokerUrl())) {
@SuppressWarnings("unchecked")
final PulsarTemplate<String> pulsarTemplate = context.getBean(PulsarTemplate.class);
pulsarTemplate.setDefaultTopicName("hello-pulsar-exclusive");
for (int i = 0; i < 10; i++) {
pulsarTemplate.send("John Doe");
pulsarTemplate.send("hello-pulsar-exclusive", "John Doe");
}
final boolean await = latch2.await(10, TimeUnit.SECONDS);
assertThat(await).isTrue();

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

@@ -34,15 +34,15 @@ public class PulsarBootApp {
@Bean
public ApplicationRunner runner(PulsarTemplate<Foo> pulsarTemplate) {
pulsarTemplate.setDefaultTopicName("hello-pulsar-exclusive-2");
String topic = "hello-pulsar-exclusive-2";
return args -> {
// for (int i = 0; i < 100; i ++) {
// pulsarTemplate.send("This is message " + (i + 1));
// pulsarTemplate.send(topic, "This is message " + (i + 1));
// }
Foo foo = new Foo();
foo.setFoo("Foo");
foo.setBar("Bar");
pulsarTemplate.send(foo);
pulsarTemplate.send(topic, foo);
};
}

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

@@ -42,13 +42,13 @@ public class ProducerApp {
@Bean
public ApplicationRunner runner(PulsarTemplate<String> pulsarTemplate) {
pulsarTemplate.setDefaultTopicName("failover-demo-topic");
String topic = "failover-demo-topic";
return args -> {
for (int i = 0; i < 100; i++) {
pulsarTemplate.sendAsync("hello john doex " + new Random().nextInt(), new FooRouter());
pulsarTemplate.sendAsync("hello alice doex " + new Random().nextInt(), new BarRouter());
pulsarTemplate.sendAsync(topic, "hello john doex " + new Random().nextInt(), new FooRouter());
pulsarTemplate.sendAsync(topic, "hello alice doex " + new Random().nextInt(), new BarRouter());
if (i % 2 == 0) {
pulsarTemplate.sendAsync("hello buzz doex " + new Random().nextInt(), new BuzzRouter());
pulsarTemplate.sendAsync(topic, "hello buzz doex " + new Random().nextInt(), new BuzzRouter());
}
Thread.sleep(5_000);
}

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

@@ -41,12 +41,12 @@ public class FailoverConsumerApp {
@Bean
public ApplicationRunner runner(PulsarTemplate<String> pulsarTemplate) {
pulsarTemplate.setDefaultTopicName("failover-demo-topic");
String topic = "failover-demo-topic";
return args -> {
for (int i = 0; i < 10; i++) {
pulsarTemplate.sendAsync("hello john doe 0 ", new FooRouter());
pulsarTemplate.sendAsync("hello alice doe 1", new BarRouter());
pulsarTemplate.sendAsync("hello buzz doe 2", new BuzzRouter());
pulsarTemplate.sendAsync(topic, "hello john doe 0 ", new FooRouter());
pulsarTemplate.sendAsync(topic, "hello alice doe 1", new BarRouter());
pulsarTemplate.sendAsync(topic, "hello buzz doe 2", new BuzzRouter());
Thread.sleep(1_000);
}
System.exit(0);

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

@@ -43,9 +43,9 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
private final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass()));
private final Map<String, Object> producerConfig = new HashMap<>();
// TODO add caching of producers per schema/topic w/ ttl
private Producer<T> producer;
private final Map<String, Object> producerConfig = new HashMap<>();
private final PulsarClient pulsarClient;
@@ -57,12 +57,13 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
}
@Override
public Producer<T> createProducer(Schema<T> schema) throws PulsarClientException {
return createProducer(schema, null);
public Producer<T> createProducer(String topic, Schema<T> schema) throws PulsarClientException {
return createProducer(topic, schema, null);
}
@Override
public Producer<T> createProducer(Schema<T> schema, MessageRouter messageRouter) throws PulsarClientException {
public Producer<T> createProducer(String topic, Schema<T> schema, MessageRouter messageRouter) throws PulsarClientException {
this.logger.trace(() -> String.format("Creating producer for '%s' topic", topic));
final ProducerBuilder<T> producerBuilder = this.pulsarClient.newProducer(schema);
if (!CollectionUtils.isEmpty(this.producerConfig)) {
producerBuilder.loadConf(this.producerConfig);
@@ -70,8 +71,10 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
if (messageRouter != null) {
producerBuilder.messageRouter(messageRouter);
}
this.producer = producerBuilder.create();
return this.producer;
if (topic != null) {
producerBuilder.topic(topic);
}
return producerBuilder.create();
}
@Override
@@ -80,8 +83,6 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
}
@Override
public void destroy() throws Exception {
this.logger.info("Closing producer");
this.producer.close();
public void destroy() {
}
}

94
spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarOperations.java Normal file
View File

@@ -0,0 +1,94 @@
/*
* 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.util.concurrent.CompletableFuture;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.MessageRouter;
import org.apache.pulsar.client.api.PulsarClientException;
/**
* The basic Pulsar operations contract.
*
* @param <T> the message payload type
*
* @author Chris Bono
*/
public interface PulsarOperations<T> {
/**
* Sends a message to the default topic in a blocking manner.
* @param message the message to send
* @return the id of the sent message
* @throws PulsarClientException if an error occurs
*/
default MessageId send(T message) throws PulsarClientException {
return send(null, message);
}
/**
* Sends a message to the specified topic in a blocking manner.
* @param topic the topic to send the message to or {@code null} to send to the default topic
* @param message the message to send
* @return the id of the sent message
* @throws PulsarClientException if an error occurs
*/
MessageId send(String topic, T message) throws PulsarClientException;
/**
* Sends a message to the default topic in a blocking manner.
* @param message the message to send
* @return a future that holds the id of the sent message
* @throws PulsarClientException if an error occurs
*/
default CompletableFuture<MessageId> sendAsync(T message) throws PulsarClientException {
return sendAsync(null, message);
}
/**
* Sends a message to the specified topic in a blocking manner.
* @param topic the topic to send the message to or {@code null} to send to the default topic
* @param message the message to send
* @return a future that holds the id of the sent message
* @throws PulsarClientException if an error occurs
*/
default CompletableFuture<MessageId> sendAsync(String topic, T message) throws PulsarClientException {
return sendAsync(topic, message, null);
}
/**
* Sends a message to the default topic in a blocking manner.
* @param message the message to send
* @param messageRouter the optional message router to use
* @return a future that holds the id of the sent message
* @throws PulsarClientException if an error occurs
*/
default CompletableFuture<MessageId> sendAsync(T message, MessageRouter messageRouter) throws PulsarClientException {
return sendAsync(null, message, messageRouter);
}
/**
* Sends a message to the specified topic in a non-blocking manner.
* @param topic the topic to send the message to or {@code null} to send to the default topic
* @param message the message to send
* @param messageRouter the optional message router to use
* @return a future that holds the id of the sent message
* @throws PulsarClientException if an error occurs
*/
CompletableFuture<MessageId> sendAsync(String topic, T message, MessageRouter messageRouter) throws PulsarClientException;
}

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

@@ -36,21 +36,23 @@ public interface PulsarProducerFactory<T> {
/**
* Create a producer.
*
* @param topic the topic the producer will send messages to or {@code null} to use the default topic
* @param schema the schema of the messages to be sent
* @return the producer
* @throws PulsarClientException if any error occurs
*/
Producer<T> createProducer(Schema<T> schema) throws PulsarClientException;
Producer<T> createProducer(String topic, Schema<T> schema) throws PulsarClientException;
/**
* Create a producer.
*
* @param topic the topic the producer will send messages to or {@code null} to use the default topic
* @param schema the schema of the messages to be sent
* @param messageRouter the optional message router to use
* @return the producer
* @throws PulsarClientException if any error occurs
*/
Producer<T> createProducer(Schema<T> schema, MessageRouter messageRouter) throws PulsarClientException;
Producer<T> createProducer(String topic, Schema<T> schema, MessageRouter messageRouter) throws PulsarClientException;
/**
* Return a map of configuration options to use when creating producers.

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

@@ -16,125 +16,88 @@
package org.springframework.pulsar.core;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.logging.LogFactory;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.MessageRouter;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.springframework.core.log.LogAccessor;
import org.springframework.util.StringUtils;
/**
* Template implementation for publishing to Pulsar topics.
* A thread-safe template for executing high-level Pulsar operations.
*
* @param <T> message type.
* @param <T> the message payload type
*
* @author Soby Chacko
* @author Chris Bono
*/
public class PulsarTemplate<T> {
public class PulsarTemplate<T> implements PulsarOperations<T> {
private final Map<SchemaTopic, Producer<T>> producerCache = new ConcurrentHashMap<>();
private final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass()));
private final PulsarProducerFactory<T> pulsarProducerFactory;
private final PulsarProducerFactory<T> producerFactory;
private String defaultTopicName;
private Schema<T> schema;
public PulsarTemplate(PulsarProducerFactory<T> pulsarProducerFactory) {
this.pulsarProducerFactory = pulsarProducerFactory;
/**
* Constructs a template instance.
* @param producerFactory the producer factory used to create the backing Pulsar producers.
*/
public PulsarTemplate(PulsarProducerFactory<T> producerFactory) {
this.producerFactory = producerFactory;
}
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 = this.producerCache.get(schemaTopic);
if (producer == null) {
producer = this.pulsarProducerFactory.createProducer(schema);
this.producerCache.put(schemaTopic, producer);
@Override
public MessageId send(String topic, T message) throws PulsarClientException {
try {
return this.sendAsync(topic, message).get();
}
return producer.send(message);
}
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 = this.producerCache.get(schemaTopic);
if (producer == null) {
producer = this.pulsarProducerFactory.createProducer(schema);
this.producerCache.put(schemaTopic, producer);
catch (Exception ex) {
throw PulsarClientException.unwrap(ex);
}
return producer.sendAsync(message);
}
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 = this.producerCache.get(schemaTopic);
if (producer == null) {
producer = this.pulsarProducerFactory.createProducer(schema, messageRouter);
this.producerCache.put(schemaTopic, producer);
@Override
public CompletableFuture<MessageId> sendAsync(String topic, T message, MessageRouter messageRouter) throws PulsarClientException {
final String topicName = resolveTopicName(topic);
this.logger.trace(() -> String.format("Sending msg to '%s' topic", topicName));
final Producer<T> producer = prepareProducerForSend(topic, message, messageRouter);
return producer.sendAsync(message)
.whenComplete((msgId, ex) -> {
if (ex == null) {
this.logger.trace(() -> String.format("Sent msg to '%s' topic", topicName));
// TODO success metrics
}
else {
this.logger.error(ex, () -> String.format("Failed to send msg to '%s' topic", topicName));
// TODO fail metrics
}
closeProducerAsync(producer);
});
}
private String resolveTopicName(String userSpecifiedTopic) {
if (StringUtils.hasText(userSpecifiedTopic)) {
return userSpecifiedTopic;
}
return producer.sendAsync(message);
return Optional.ofNullable(this.producerFactory.getProducerConfig().get("topicName"))
.map(Object::toString)
.orElseThrow(() -> new IllegalArgumentException("Topic must be specified when no default topic is configured"));
}
private SchemaTopic getSchemaTopic(Schema<T> schema, PulsarProducerFactory<T> pulsarProducerFactory, MessageRouter messageRouter) {
return new SchemaTopic(schema, (String) pulsarProducerFactory.getProducerConfig().get("topicName"), messageRouter);
private Producer<T> prepareProducerForSend(String topic, T message, MessageRouter messageRouter) throws PulsarClientException {
Schema<T> schema = SchemaUtils.getSchema(message);
return this.producerFactory.createProducer(topic, schema, messageRouter);
}
public void setDefaultTopicName(String defaultTopicName) {
this.defaultTopicName = defaultTopicName;
this.pulsarProducerFactory.getProducerConfig().put("topicName", defaultTopicName);
}
public Schema<T> getSchema() {
return this.schema;
}
public void setSchema(Schema<T> schema) {
this.schema = schema;
}
private class SchemaTopic {
final Schema<T> schema;
final String topicName;
final MessageRouter messageRouter;
SchemaTopic(Schema<T> schema, String topicName, MessageRouter messageRouter) {
this.schema = schema;
this.topicName = topicName;
this.messageRouter = messageRouter;
}
@Override
public boolean equals(Object o) {
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 this.schema.equals(that.schema) && this.topicName.equals(that.topicName);
}
else if (this.messageRouter == null) {
return false;
}
else if (that.messageRouter == null) {
return false;
}
return this.schema.equals(that.schema) && this.topicName.equals(that.topicName) && this.messageRouter.equals(that.messageRouter);
}
@Override
public int hashCode() {
return Objects.hash(this.schema, this.topicName, this.messageRouter);
}
private void closeProducerAsync(Producer<T> producer) {
producer.closeAsync().exceptionally(e -> {
this.logger.warn(e, () -> String.format("Failed to close producer %s:%s", producer.getProducerName(), producer.getTopic()));
return null;
});
}
}