diff --git a/spring-pulsar-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarListenerTests.java b/spring-pulsar-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarListenerTests.java index 22529bee..9dbd35e6 100644 --- a/spring-pulsar-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarListenerTests.java +++ b/spring-pulsar-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarListenerTests.java @@ -51,8 +51,7 @@ class PulsarListenerTests extends AbstractContainerBaseTests { try (ConfigurableApplicationContext context = app.run("--spring.pulsar.client.serviceUrl=" + AbstractContainerBaseTests.getPulsarBrokerUrl())) { @SuppressWarnings("unchecked") final PulsarTemplate 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 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(); diff --git a/spring-pulsar-sample-apps/src/main/java/app1/PulsarBootApp.java b/spring-pulsar-sample-apps/src/main/java/app1/PulsarBootApp.java index 0a5c6496..2249968e 100644 --- a/spring-pulsar-sample-apps/src/main/java/app1/PulsarBootApp.java +++ b/spring-pulsar-sample-apps/src/main/java/app1/PulsarBootApp.java @@ -34,15 +34,15 @@ public class PulsarBootApp { @Bean public ApplicationRunner runner(PulsarTemplate 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); }; } diff --git a/spring-pulsar-sample-apps/src/main/java/app2/ProducerApp.java b/spring-pulsar-sample-apps/src/main/java/app2/ProducerApp.java index d1f29726..0750bec3 100644 --- a/spring-pulsar-sample-apps/src/main/java/app2/ProducerApp.java +++ b/spring-pulsar-sample-apps/src/main/java/app2/ProducerApp.java @@ -42,13 +42,13 @@ public class ProducerApp { @Bean public ApplicationRunner runner(PulsarTemplate 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); } diff --git a/spring-pulsar-sample-apps/src/main/java/app4/FailoverConsumerApp.java b/spring-pulsar-sample-apps/src/main/java/app4/FailoverConsumerApp.java index ea178001..8ae725be 100644 --- a/spring-pulsar-sample-apps/src/main/java/app4/FailoverConsumerApp.java +++ b/spring-pulsar-sample-apps/src/main/java/app4/FailoverConsumerApp.java @@ -41,12 +41,12 @@ public class FailoverConsumerApp { @Bean public ApplicationRunner runner(PulsarTemplate 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); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java index b602bba1..dbfc047d 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java @@ -43,9 +43,9 @@ public class DefaultPulsarProducerFactory implements PulsarProducerFactory private final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass())); - private final Map producerConfig = new HashMap<>(); + // TODO add caching of producers per schema/topic w/ ttl - private Producer producer; + private final Map producerConfig = new HashMap<>(); private final PulsarClient pulsarClient; @@ -57,12 +57,13 @@ public class DefaultPulsarProducerFactory implements PulsarProducerFactory } @Override - public Producer createProducer(Schema schema) throws PulsarClientException { - return createProducer(schema, null); + public Producer createProducer(String topic, Schema schema) throws PulsarClientException { + return createProducer(topic, schema, null); } @Override - public Producer createProducer(Schema schema, MessageRouter messageRouter) throws PulsarClientException { + public Producer createProducer(String topic, Schema schema, MessageRouter messageRouter) throws PulsarClientException { + this.logger.trace(() -> String.format("Creating producer for '%s' topic", topic)); final ProducerBuilder producerBuilder = this.pulsarClient.newProducer(schema); if (!CollectionUtils.isEmpty(this.producerConfig)) { producerBuilder.loadConf(this.producerConfig); @@ -70,8 +71,10 @@ public class DefaultPulsarProducerFactory implements PulsarProducerFactory 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 implements PulsarProducerFactory } @Override - public void destroy() throws Exception { - this.logger.info("Closing producer"); - this.producer.close(); + public void destroy() { } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarOperations.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarOperations.java new file mode 100644 index 00000000..6b58fa38 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarOperations.java @@ -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 the message payload type + * + * @author Chris Bono + */ +public interface PulsarOperations { + + /** + * 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 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 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 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 sendAsync(String topic, T message, MessageRouter messageRouter) throws PulsarClientException; +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java index d43036b7..a4a4704f 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java @@ -36,21 +36,23 @@ public interface PulsarProducerFactory { /** * 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 createProducer(Schema schema) throws PulsarClientException; + Producer createProducer(String topic, Schema 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 createProducer(Schema schema, MessageRouter messageRouter) throws PulsarClientException; + Producer createProducer(String topic, Schema schema, MessageRouter messageRouter) throws PulsarClientException; /** * Return a map of configuration options to use when creating producers. diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java index acfb2fca..84fe14c7 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java @@ -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 message type. + * @param the message payload type * * @author Soby Chacko + * @author Chris Bono */ -public class PulsarTemplate { +public class PulsarTemplate implements PulsarOperations { - private final Map> producerCache = new ConcurrentHashMap<>(); + private final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass())); - private final PulsarProducerFactory pulsarProducerFactory; + private final PulsarProducerFactory producerFactory; - private String defaultTopicName; - - private Schema schema; - - public PulsarTemplate(PulsarProducerFactory pulsarProducerFactory) { - this.pulsarProducerFactory = pulsarProducerFactory; + /** + * Constructs a template instance. + * @param producerFactory the producer factory used to create the backing Pulsar producers. + */ + public PulsarTemplate(PulsarProducerFactory producerFactory) { + this.producerFactory = producerFactory; } - public MessageId send(T message) throws PulsarClientException { - final Schema schema = this.schema != null ? this.schema : SchemaUtils.getSchema(message); - final SchemaTopic schemaTopic = getSchemaTopic(schema, this.pulsarProducerFactory, null); - Producer producer = 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 sendAsync(T message) throws PulsarClientException { - final Schema schema = this.schema != null ? this.schema : SchemaUtils.getSchema(message); - final SchemaTopic schemaTopic = getSchemaTopic(schema, this.pulsarProducerFactory, null); - Producer producer = 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 sendAsync(T message, MessageRouter messageRouter) throws PulsarClientException { - final Schema schema = this.schema != null ? this.schema : SchemaUtils.getSchema(message); - final SchemaTopic schemaTopic = getSchemaTopic(schema, this.pulsarProducerFactory, messageRouter); - Producer producer = this.producerCache.get(schemaTopic); - if (producer == null) { - producer = this.pulsarProducerFactory.createProducer(schema, messageRouter); - this.producerCache.put(schemaTopic, producer); + @Override + public CompletableFuture 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 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 schema, PulsarProducerFactory pulsarProducerFactory, MessageRouter messageRouter) { - return new SchemaTopic(schema, (String) pulsarProducerFactory.getProducerConfig().get("topicName"), messageRouter); + private Producer prepareProducerForSend(String topic, T message, MessageRouter messageRouter) throws PulsarClientException { + Schema 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 getSchema() { - return this.schema; - } - - public void setSchema(Schema schema) { - this.schema = schema; - } - - private class SchemaTopic { - - final Schema schema; - final String topicName; - final MessageRouter messageRouter; - - SchemaTopic(Schema 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 producer) { + producer.closeAsync().exceptionally(e -> { + this.logger.warn(e, () -> String.format("Failed to close producer %s:%s", producer.getProducerName(), producer.getTopic())); + return null; + }); } }