Producer and sender factories use topic resolver (#322)

* Producer and sender factories use topic resolver

* Remove resolveTopicName from ProducerUtils
* DefaultPulsarProducerFactory and CachingPulsarProducerFactory
  use TopicResolver instead of ProducerUtils to resolve topic
  name.
* Delete ReactiveMessageSenderUtils
* DefaultReactivePulsarSenderFactory use TopicResolver instead of
  ReactiveMessageSenderUtils to resolve topic name

* Disabled shared consumer test again

spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java

@@ -33,6 +33,8 @@ import org.apache.pulsar.reactive.client.api.ReactivePulsarClient;
 
 import org.springframework.core.log.LogAccessor;
 import org.springframework.lang.Nullable;
+import org.springframework.pulsar.core.DefaultTopicResolver;
+import org.springframework.pulsar.core.TopicResolver;
 import org.springframework.util.CollectionUtils;
 
 /**
@@ -53,20 +55,23 @@ public class DefaultReactivePulsarSenderFactory<T> implements ReactivePulsarSend
 	@Nullable
 	private final ReactiveMessageSenderCache reactiveMessageSenderCache;
 
+	private TopicResolver topicResolver;
+
 	public DefaultReactivePulsarSenderFactory(PulsarClient pulsarClient,
 			@Nullable ReactiveMessageSenderSpec reactiveMessageSenderSpec,
 			@Nullable ReactiveMessageSenderCache reactiveMessageSenderCache) {
 		this(AdaptedReactivePulsarClientFactory.create(pulsarClient), reactiveMessageSenderSpec,
-				reactiveMessageSenderCache);
+				reactiveMessageSenderCache, new DefaultTopicResolver());
 	}
 
 	public DefaultReactivePulsarSenderFactory(ReactivePulsarClient reactivePulsarClient,
 			@Nullable ReactiveMessageSenderSpec reactiveMessageSenderSpec,
-			@Nullable ReactiveMessageSenderCache reactiveMessageSenderCache) {
+			@Nullable ReactiveMessageSenderCache reactiveMessageSenderCache, TopicResolver topicResolver) {
 		this.reactivePulsarClient = reactivePulsarClient;
 		this.reactiveMessageSenderSpec = new ImmutableReactiveMessageSenderSpec(
 				reactiveMessageSenderSpec != null ? reactiveMessageSenderSpec : new MutableReactiveMessageSenderSpec());
 		this.reactiveMessageSenderCache = reactiveMessageSenderCache;
+		this.topicResolver = topicResolver;
 	}
 
 	@Override
@@ -90,8 +95,12 @@ public class DefaultReactivePulsarSenderFactory<T> implements ReactivePulsarSend
 	private ReactiveMessageSender<T> doCreateReactiveMessageSender(Schema<T> schema, @Nullable String topic,
 			@Nullable List<ReactiveMessageSenderBuilderCustomizer<T>> customizers) {
 		Objects.requireNonNull(schema, "Schema must be specified");
-		String resolvedTopic = ReactiveMessageSenderUtils.resolveTopicName(topic, this);
+		String resolvedTopic = this.topicResolver
+				.resolveTopic(topic, () -> getReactiveMessageSenderSpec().getTopicName())
+				.orElseThrow(() -> new IllegalArgumentException(
+						"Topic must be specified when no default topic is configured"));
 		this.logger.trace(() -> "Creating reactive message sender for '%s' topic".formatted(resolvedTopic));
+
 		ReactiveMessageSenderBuilder<T> sender = this.reactivePulsarClient.messageSender(schema);
 		sender.applySpec(this.reactiveMessageSenderSpec);
 		sender.topic(resolvedTopic);

spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactiveMessageSenderUtils.java

@@ -1,47 +0,0 @@
-/*
- * Copyright 2022-2023 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.reactive.core;
-
-import java.util.Optional;
-
-import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderSpec;
-
-import org.springframework.lang.Nullable;
-import org.springframework.util.StringUtils;
-
-/**
- * Common utilities used by reactive sender components.
- *
- * @author Christophe Bornet
- */
-final class ReactiveMessageSenderUtils {
-
-	private ReactiveMessageSenderUtils() {
-	}
-
-	static <T> String resolveTopicName(@Nullable String userSpecifiedTopic,
-			ReactivePulsarSenderFactory<T> reactiveMessageSenderFactory) {
-		ReactiveMessageSenderSpec reactiveMessageSenderSpec = reactiveMessageSenderFactory
-				.getReactiveMessageSenderSpec();
-		if (StringUtils.hasText(userSpecifiedTopic)) {
-			return userSpecifiedTopic;
-		}
-		return Optional.ofNullable(reactiveMessageSenderSpec).map(ReactiveMessageSenderSpec::getTopicName).orElseThrow(
-				() -> new IllegalArgumentException("Topic must be specified when no default topic is configured"));
-	}
-
-}

spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/DefaultReactivePulsarMessageListenerContainerTests.java

@@ -39,6 +39,7 @@ import org.apache.pulsar.reactive.client.api.ReactivePulsarClient;
 import org.assertj.core.api.InstanceOfAssertFactories;
 import org.junit.jupiter.api.Test;
 
+import org.springframework.pulsar.core.DefaultTopicResolver;
 import org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory;
 import org.springframework.pulsar.reactive.core.DefaultReactivePulsarSenderFactory;
 import org.springframework.pulsar.reactive.core.ReactivePulsarTemplate;
@@ -79,7 +80,7 @@ class DefaultReactivePulsarMessageListenerContainerTests implements PulsarTestCo
 		MutableReactiveMessageSenderSpec prodConfig = new MutableReactiveMessageSenderSpec();
 		prodConfig.setTopicName(topic);
 		DefaultReactivePulsarSenderFactory<String> pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>(
-				reactivePulsarClient, prodConfig, null);
+				reactivePulsarClient, prodConfig, null, new DefaultTopicResolver());
 		ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(pulsarProducerFactory);
 		pulsarTemplate.send("hello john doe").subscribe();
 		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -111,7 +112,7 @@ class DefaultReactivePulsarMessageListenerContainerTests implements PulsarTestCo
 		MutableReactiveMessageSenderSpec prodConfig = new MutableReactiveMessageSenderSpec();
 		prodConfig.setTopicName(topic);
 		DefaultReactivePulsarSenderFactory<String> pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>(
-				reactivePulsarClient, prodConfig, null);
+				reactivePulsarClient, prodConfig, null, new DefaultTopicResolver());
 		ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(pulsarProducerFactory);
 		Flux.range(0, 5).map(i -> "hello john doe" + i).as(pulsarTemplate::send).subscribe();
 		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -150,7 +151,7 @@ class DefaultReactivePulsarMessageListenerContainerTests implements PulsarTestCo
 		MutableReactiveMessageSenderSpec prodConfig = new MutableReactiveMessageSenderSpec();
 		prodConfig.setTopicName(topic);
 		DefaultReactivePulsarSenderFactory<String> pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>(
-				reactivePulsarClient, prodConfig, null);
+				reactivePulsarClient, prodConfig, null, new DefaultTopicResolver());
 		ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(pulsarProducerFactory);
 		pulsarTemplate.send("hello john doe").subscribe();
 		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -259,7 +260,7 @@ class DefaultReactivePulsarMessageListenerContainerTests implements PulsarTestCo
 		MutableReactiveMessageSenderSpec prodConfig = new MutableReactiveMessageSenderSpec();
 		prodConfig.setTopicName(topic);
 		DefaultReactivePulsarSenderFactory<String> pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>(
-				reactivePulsarClient, prodConfig, null);
+				reactivePulsarClient, prodConfig, null, new DefaultTopicResolver());
 		ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(pulsarProducerFactory);
 		pulsarTemplate.send("hello john doe").subscribe();
 		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -305,7 +306,7 @@ class DefaultReactivePulsarMessageListenerContainerTests implements PulsarTestCo
 		prodConfig.setBatchingEnabled(false);
 		prodConfig.setTopicName(topic);
 		DefaultReactivePulsarSenderFactory<String> pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>(
-				reactivePulsarClient, prodConfig, null);
+				reactivePulsarClient, prodConfig, null, new DefaultTopicResolver());
 		ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(pulsarProducerFactory);
 		Flux.range(0, 5).map(i -> "hello john doe" + i).as(pulsarTemplate::send).subscribe();
 		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();

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

@@ -86,16 +86,18 @@ public class PulsarAutoConfiguration {
 	@Bean
 	@ConditionalOnMissingBean
 	@ConditionalOnProperty(name = "spring.pulsar.producer.cache.enabled", havingValue = "false")
-	public PulsarProducerFactory<?> pulsarProducerFactory(PulsarClient pulsarClient) {
-		return new DefaultPulsarProducerFactory<>(pulsarClient, this.properties.buildProducerProperties());
+	public PulsarProducerFactory<?> pulsarProducerFactory(PulsarClient pulsarClient, TopicResolver topicResolver) {
+		return new DefaultPulsarProducerFactory<>(pulsarClient, this.properties.buildProducerProperties(),
+				topicResolver);
 	}
 
 	@Bean
 	@ConditionalOnMissingBean
 	@ConditionalOnProperty(name = "spring.pulsar.producer.cache.enabled", havingValue = "true", matchIfMissing = true)
-	public PulsarProducerFactory<?> cachingPulsarProducerFactory(PulsarClient pulsarClient) {
+	public PulsarProducerFactory<?> cachingPulsarProducerFactory(PulsarClient pulsarClient,
+			TopicResolver topicResolver) {
 		return new CachingPulsarProducerFactory<>(pulsarClient, this.properties.buildProducerProperties(),
-				this.properties.getProducer().getCache().getExpireAfterAccess(),
+				topicResolver, this.properties.getProducer().getCache().getExpireAfterAccess(),
 				this.properties.getProducer().getCache().getMaximumSize(),
 				this.properties.getProducer().getCache().getInitialCapacity());
 	}

spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAutoConfiguration.java

@@ -93,9 +93,9 @@ public class PulsarReactiveAutoConfiguration {
 	@Bean
 	@ConditionalOnMissingBean
 	public ReactivePulsarSenderFactory<?> reactivePulsarSenderFactory(ReactivePulsarClient pulsarReactivePulsarClient,
-			ObjectProvider<ReactiveMessageSenderCache> cache) {
+			ObjectProvider<ReactiveMessageSenderCache> cache, TopicResolver topicResolver) {
 		return new DefaultReactivePulsarSenderFactory<>(pulsarReactivePulsarClient,
-				this.properties.buildReactiveMessageSenderSpec(), cache.getIfAvailable());
+				this.properties.buildReactiveMessageSenderSpec(), cache.getIfAvailable(), topicResolver);
 	}
 
 	@Bean

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

@@ -22,6 +22,7 @@ import static org.mockito.Mockito.mock;
 
 import java.util.concurrent.TimeUnit;
 
+import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.SubscriptionInitialPosition;
 import org.apache.pulsar.client.api.SubscriptionType;
@@ -465,6 +466,22 @@ class PulsarAutoConfigurationTests {
 							.hasFieldOrPropertyWithValue("expiresAfterAccessNanos", TimeUnit.SECONDS.toNanos(100))));
 		}
 
+		@Test
+		void beansAreInjectedInNonCachingProducerFactory() {
+			contextRunner.withPropertyValues("spring.pulsar.producer.cache.enabled=false")
+					.run((context -> assertThat(context).hasNotFailed().getBean(DefaultPulsarProducerFactory.class)
+							.hasFieldOrPropertyWithValue("pulsarClient", context.getBean(PulsarClient.class))
+							.hasFieldOrPropertyWithValue("topicResolver", context.getBean(TopicResolver.class))));
+		}
+
+		@Test
+		void beansAreInjectedInCachingProducerFactory() {
+			contextRunner.withPropertyValues("spring.pulsar.producer.cache.enabled=true")
+					.run((context -> assertThat(context).hasNotFailed().getBean(CachingPulsarProducerFactory.class)
+							.hasFieldOrPropertyWithValue("pulsarClient", context.getBean(PulsarClient.class))
+							.hasFieldOrPropertyWithValue("topicResolver", context.getBean(TopicResolver.class))));
+		}
+
 		private void assertHasProducerFactoryOfType(Class<?> producerFactoryType,
 				AssertableApplicationContext context) {
 			assertThat(context).hasNotFailed().hasSingleBean(PulsarProducerFactory.class)

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

@@ -50,6 +50,7 @@ import org.springframework.boot.test.context.assertj.AssertableApplicationContex
 import org.springframework.boot.test.context.runner.ApplicationContextRunner;
 import org.springframework.pulsar.config.PulsarClientFactoryBean;
 import org.springframework.pulsar.core.SchemaResolver;
+import org.springframework.pulsar.core.TopicResolver;
 import org.springframework.pulsar.reactive.config.DefaultReactivePulsarListenerContainerFactory;
 import org.springframework.pulsar.reactive.config.ReactivePulsarListenerContainerFactory;
 import org.springframework.pulsar.reactive.config.ReactivePulsarListenerEndpointRegistry;
@@ -69,6 +70,7 @@ import org.springframework.pulsar.reactive.listener.ReactivePulsarContainerPrope
  * Autoconfiguration tests for {@link PulsarReactiveAutoConfiguration}.
  *
  * @author Christophe Bornet
+ * @author Chris Bono
  */
 @SuppressWarnings("unchecked")
 class PulsarReactiveAutoConfigurationTests {
@@ -201,6 +203,9 @@ class PulsarReactiveAutoConfigurationTests {
 								.extracting("reactiveMessageSenderCache",
 										InstanceOfAssertFactories.type(ReactiveMessageSenderCache.class))
 								.isSameAs(cache);
+						senderFactory.extracting("topicResolver", InstanceOfAssertFactories.type(TopicResolver.class))
+								.isSameAs(context.getBean(TopicResolver.class));
+
 					}));
 		}
 	}

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

@@ -73,13 +73,15 @@ public class CachingPulsarProducerFactory<T> extends DefaultPulsarProducerFactor
 	 * configuration.
 	 * @param pulsarClient the client used to create the producers
 	 * @param producerConfig the configuration to use when creating a producer
+	 * @param topicResolver the topic resolver to use
 	 * @param cacheExpireAfterAccess time period to expire unused entries in the cache
 	 * @param cacheMaximumSize maximum size of cache (entries)
 	 * @param cacheInitialCapacity the initial size of cache
 	 */
 	public CachingPulsarProducerFactory(PulsarClient pulsarClient, Map<String, Object> producerConfig,
-			Duration cacheExpireAfterAccess, Long cacheMaximumSize, Integer cacheInitialCapacity) {
-		super(pulsarClient, producerConfig);
+			TopicResolver topicResolver, Duration cacheExpireAfterAccess, Long cacheMaximumSize,
+			Integer cacheInitialCapacity) {
+		super(pulsarClient, producerConfig, topicResolver);
 		this.producerCache = Caffeine.newBuilder().expireAfterAccess(cacheExpireAfterAccess)
 				.maximumSize(cacheMaximumSize).initialCapacity(cacheInitialCapacity)
 				.scheduler(Scheduler.systemScheduler()).evictionListener(
@@ -95,7 +97,7 @@ public class CachingPulsarProducerFactory<T> extends DefaultPulsarProducerFactor
 	protected Producer<T> doCreateProducer(Schema<T> schema, @Nullable String topic,
 			@Nullable Collection<String> encryptionKeys, @Nullable List<ProducerBuilderCustomizer<T>> customizers) {
 		Objects.requireNonNull(schema, "Schema must be specified");
-		String resolveTopicName = ProducerUtils.resolveTopicName(topic, this);
+		String resolveTopicName = resolveTopicName(topic);
 		ProducerCacheKey<T> producerCacheKey = new ProducerCacheKey<>(schema, resolveTopicName,
 				encryptionKeys == null ? null : new HashSet<>(encryptionKeys), customizers);
 		return this.producerCache.get(producerCacheKey,

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

@@ -53,9 +53,17 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
 
 	private final PulsarClient pulsarClient;
 
+	private final TopicResolver topicResolver;
+
 	public DefaultPulsarProducerFactory(PulsarClient pulsarClient, Map<String, Object> config) {
+		this(pulsarClient, config, new DefaultTopicResolver());
+	}
+
+	public DefaultPulsarProducerFactory(PulsarClient pulsarClient, Map<String, Object> config,
+			TopicResolver topicResolver) {
 		this.pulsarClient = pulsarClient;
 		this.producerConfig = Collections.unmodifiableMap(config);
+		this.topicResolver = topicResolver;
 	}
 
 	@Override
@@ -94,7 +102,7 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
 			@Nullable Collection<String> encryptionKeys, @Nullable List<ProducerBuilderCustomizer<T>> customizers)
 			throws PulsarClientException {
 		Objects.requireNonNull(schema, "Schema must be specified");
-		String resolvedTopic = ProducerUtils.resolveTopicName(topic, this);
+		String resolvedTopic = resolveTopicName(topic);
 		this.logger.trace(() -> "Creating producer for '%s' topic".formatted(resolvedTopic));
 		ProducerBuilder<T> producerBuilder = this.pulsarClient.newProducer(schema);
 
@@ -116,6 +124,12 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
 		return producerBuilder.create();
 	}
 
+	protected String resolveTopicName(String userSpecifiedTopic) {
+		String defaultTopic = Objects.toString(getProducerConfig().get("topicName"), null);
+		return this.topicResolver.resolveTopic(userSpecifiedTopic, () -> defaultTopic).orElseThrow(
+				() -> new IllegalArgumentException("Topic must be specified when no default topic is configured"));
+	}
+
 	@Override
 	public Map<String, Object> getProducerConfig() {
 		return this.producerConfig;

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

@@ -16,13 +16,9 @@
 
 package org.springframework.pulsar.core;
 
-import java.util.Optional;
-
 import org.apache.pulsar.client.api.Producer;
 
 import org.springframework.core.log.LogAccessor;
-import org.springframework.lang.Nullable;
-import org.springframework.util.StringUtils;
 
 /**
  * Common utilities used by producer components.
@@ -38,15 +34,6 @@ final class ProducerUtils {
 		return "(%s:%s)".formatted(producer.getProducerName(), producer.getTopic());
 	}
 
-	static <T> String resolveTopicName(@Nullable String userSpecifiedTopic, PulsarProducerFactory<T> producerFactory) {
-		if (StringUtils.hasText(userSpecifiedTopic)) {
-			return userSpecifiedTopic;
-		}
-		return Optional.ofNullable(producerFactory.getProducerConfig().get("topicName")).map(Object::toString)
-				.orElseThrow(() -> new IllegalArgumentException(
-						"Topic must be specified when no default topic is configured"));
-	}
-
 	static <T> void closeProducerAsync(Producer<T> producer, LogAccessor logger) {
 		producer.closeAsync().exceptionally(e -> {
 			logger.warn(e, () -> "Failed to close producer %s".formatted(ProducerUtils.formatProducer(producer)));

spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java

@@ -186,7 +186,7 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests {
 	@Test
 	void producerEvictedFromCache() throws PulsarClientException {
 		CachingPulsarProducerFactory<String> producerFactory = new CachingPulsarProducerFactory<>(pulsarClient,
-				Collections.emptyMap(), Duration.ofSeconds(3L), 10L, 2);
+				Collections.emptyMap(), new DefaultTopicResolver(), Duration.ofSeconds(3L), 10L, 2);
 		var actualProducer = actualProducer(producerFactory.createProducer(schema, "topic1"));
 		var cacheKey = new ProducerCacheKey<>(schema, "topic1", null, null);
 		var producerCache = getAssertedProducerCache(producerFactory, Collections.singletonList(cacheKey));
@@ -232,7 +232,7 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests {
 	protected CachingPulsarProducerFactory<String> producerFactory(PulsarClient pulsarClient,
 			Map<String, Object> producerConfig) {
 		var producerFactory = new CachingPulsarProducerFactory<String>(pulsarClient, producerConfig,
-				Duration.ofMinutes(5L), 30L, 2);
+				new DefaultTopicResolver(), Duration.ofMinutes(5L), 30L, 2);
 		producerFactories.add(producerFactory);
 		return producerFactory;
 	}

spring-pulsar/src/test/java/org/springframework/pulsar/core/SharedSubscriptionConsumerTests.java

@@ -27,6 +27,7 @@ import java.util.concurrent.TimeUnit;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.SubscriptionType;
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.core.log.LogAccessor;
@@ -46,6 +47,7 @@ public class SharedSubscriptionConsumerTests implements PulsarTestContainerSuppo
 	private final LogAccessor logger = new LogAccessor(this.getClass());
 
 	@Test
+	@Disabled
 	void sharedSubscriptionRoundRobinBasicScenario() throws Exception {
 
 		DefaultPulsarMessageListenerContainer<String> container1 = null;