Refactor template/producer to not use default topic name

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

@@ -51,8 +51,7 @@ void testBasicListener() throws Exception {
 		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 @@ void testBatchListener() throws Exception {
 		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();

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

@@ -34,15 +34,15 @@ public static void main(String[] args) {
 
 	@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);
 
 		};
 	}

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

@@ -42,13 +42,13 @@ public static void main(String[] args) {
 
 	@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);
 			}

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

@@ -41,12 +41,12 @@ public static void main(String[] args) {
 
 	@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);

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

@@ -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,21 +57,24 @@ public DefaultPulsarProducerFactory(PulsarClient pulsarClient, Map<String, Objec
 	}
 
 	@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);
 		}
 		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 Map<String, Object> getProducerConfig() {
 	}
 
 	@Override
-	public void destroy() throws Exception {
-		this.logger.info("Closing producer");
-		this.producer.close();
+	public void destroy() {
 	}
 }

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

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

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