Add support for SpEL topics on @PulsarMessage

This commit adds support for expressions (property placeholders and
SpEL) on the topic attribute of @PulsarMessage.

Resolves #568

Author: jonas-grgt

integration-tests/src/intTest/java/org/springframework/pulsar/inttest/app/TopicPropertyDefinedSampleMessage.java

@@ -0,0 +1,23 @@
+/*
+ * Copyright 2012-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.inttest.app;
+
+import org.springframework.pulsar.annotation.PulsarMessage;
+
+@PulsarMessage(topic = "${inttest.topic}")
+record TopicPropertyDefinedSampleMessage(Integer id, String content) {
+}

integration-tests/src/intTest/java/org/springframework/pulsar/inttest/app/TopicSpELDefinedSampleMessage.java

@@ -0,0 +1,23 @@
+/*
+ * Copyright 2012-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.inttest.app;
+
+import org.springframework.pulsar.annotation.PulsarMessage;
+
+@PulsarMessage(topic = "#{spelTopic}")
+record TopicSpELDefinedSampleMessage(Integer id, String content) {
+}

spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/topic-resolution.adoc

@@ -48,6 +48,18 @@ record Foo(String value) {
 }
 ----
 
+Property placeholders and SpEL expressions are supported in the `@PulsarMessage` annotation,
+for example:
+[source,java,indent=0,subs="verbatim"]
+----
+@PulsarMessage(topic = "${app.topics.foo}")
+record Foo(String value) {
+}
+
+@PulsarMessage(topic = "#{someBean.getTopic()}")
+record Bar(String value) {
+}
+----
 === Custom topic resolver
 The preferred method of adding mappings is via the property mentioned above.
 However, if more control is needed you can replace the default resolver by proving your own implementation, for example:

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

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2012-2024 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.springframework.beans.factory.config.BeanExpressionContext;
+import org.springframework.beans.factory.config.BeanExpressionResolver;
+import org.springframework.beans.factory.config.ConfigurableBeanFactory;
+
+/**
+ * Default implementation of {@link ExpressionResolver} that relies on the
+ * {@link ConfigurableBeanFactory} capabilities to resolve expressions.
+ *
+ * @author Jonas Geiregat
+ */
+public class DefaultExpressionResolver implements ExpressionResolver {
+
+	private final BeanExpressionResolver beanExpressionResolver;
+
+	private final BeanExpressionContext beanExpressionContext;
+
+	private final ConfigurableBeanFactory configurableBeanFactory;
+
+	public DefaultExpressionResolver(ConfigurableBeanFactory configurableBeanFactory) {
+		this.beanExpressionResolver = configurableBeanFactory.getBeanExpressionResolver();
+		this.beanExpressionContext = new BeanExpressionContext(configurableBeanFactory, null);
+		this.configurableBeanFactory = configurableBeanFactory;
+	}
+
+	@Override
+	public Resolved<String> resolveToString(String expression) {
+		String placeholdersResolved = this.configurableBeanFactory.resolveEmbeddedValue(expression);
+		Object resolvedObj = this.beanExpressionResolver.evaluate(placeholdersResolved, this.beanExpressionContext);
+		if (resolvedObj instanceof String resolvedString) {
+			return Resolved.of(resolvedString);
+		}
+		if (resolvedObj == null) {
+			return Resolved.of(null);
+		}
+		return Resolved
+				.failed("The expression '%s' must resolve to a string but was: %s".formatted(expression, resolvedObj));
+	}
+
+}

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

@@ -21,6 +21,11 @@
 import java.util.Map;
 import java.util.function.Supplier;
 
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.beans.factory.config.ConfigurableBeanFactory;
+import org.springframework.core.log.LogAccessor;
 import org.springframework.lang.Nullable;
 import org.springframework.pulsar.annotation.PulsarMessage;
 import org.springframework.util.StringUtils;
@@ -34,15 +39,35 @@
  *
  * @author Chris Bono
  * @author Aleksei Arsenev
+ * @author Jonas Geiregat
  */
-public class DefaultTopicResolver implements TopicResolver {
+public class DefaultTopicResolver implements TopicResolver, BeanFactoryAware {
+
+	private final LogAccessor logger = new LogAccessor(this.getClass());
 
 	private final Map<Class<?>, String> customTopicMappings = new LinkedHashMap<>();
 
 	private final PulsarMessageAnnotationRegistry pulsarMessageAnnotationRegistry = new PulsarMessageAnnotationRegistry();
 
 	private boolean usePulsarMessageAnnotations = true;
 
+	@Nullable
+	private ExpressionResolver expressionResolver;
+
+	/**
+	 * Constructs a new DefaultTopicResolver with the given expression resolver.
+	 * @param expressionResolver the expression resolver to use for resolving topic
+	 */
+	public DefaultTopicResolver(ExpressionResolver expressionResolver) {
+		this.expressionResolver = expressionResolver;
+	}
+
+	/**
+	 * Constructs a new DefaultTopicResolver.
+	 */
+	public DefaultTopicResolver() {
+	}
+
 	/**
 	 * Sets whether to inspect message classes for the
 	 * {@link PulsarMessage @PulsarMessage} annotation during topic resolution.
@@ -135,11 +160,31 @@ protected Resolved<String> doResolveTopic(@Nullable String userSpecifiedTopic, @
 	}
 
 	// VisibleForTesting
+	@Nullable
 	String getAnnotatedTopicInfo(Class<?> messageType) {
 		return this.pulsarMessageAnnotationRegistry.getAnnotationFor(messageType)
 			.map(PulsarMessage::topic)
 			.filter(StringUtils::hasText)
+			.map(this::resolveExpression)
 			.orElse(null);
 	}
 
+	private String resolveExpression(String v) {
+		return this.expressionResolver == null ? v : this.expressionResolver.resolveToString(v)
+			.orElseThrow(() -> "Failed to resolve topic expression: %s".formatted(v));
+	}
+
+	@Override
+	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
+		if (beanFactory instanceof ConfigurableBeanFactory configurableBeanFactory) {
+			this.expressionResolver = new DefaultExpressionResolver(configurableBeanFactory);
+		}
+		else {
+			this.logger.warn(
+					() -> "Topic expressions on @PulsarMessage will not be resolved: bean factory must be %s but was %s"
+						.formatted(ConfigurableBeanFactory.class.getSimpleName(),
+								beanFactory.getClass().getSimpleName()));
+		}
+	}
+
 }

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

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2012-2024 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;
+
+/**
+ * Resolves expressions.
+ *
+ * @author Jonas Geiregat
+ */
+public interface ExpressionResolver {
+
+	/**
+	 * Resolve the given expression.
+	 * @param expression the expression to resolve
+	 * @return the resolved value as a {@code Resolved} of {@link String}
+	 */
+	Resolved<String> resolveToString(String expression);
+
+}

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

@@ -16,7 +16,9 @@
 
 package org.springframework.pulsar.core;
 
+import java.util.Objects;
 import java.util.Optional;
+import java.util.StringJoiner;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 
@@ -28,6 +30,7 @@
  * @param <T> the resolved type
  * @author Christophe Bornet
  * @author Chris Bono
+ * @author Jonas Geiregat
  */
 public final class Resolved<T> {
 
@@ -48,7 +51,7 @@ private Resolved(@Nullable T value, @Nullable RuntimeException exception) {
 	 * @param <T> the type of the value
 	 * @return a {@code Resolved} containing the resolved value
 	 */
-	public static <T> Resolved<T> of(T value) {
+	public static <T> Resolved<T> of(@Nullable T value) {
 		return new Resolved<>(value, null);
 	}
 
@@ -155,4 +158,28 @@ public T orElseThrow(Supplier<String> wrappingErrorMessage) {
 		return this.value;
 	}
 
+	@Override
+	public boolean equals(Object o) {
+		if (this == o) {
+			return true;
+		}
+		if (o == null || getClass() != o.getClass()) {
+			return false;
+		}
+		Resolved<?> resolved = (Resolved<?>) o;
+		return Objects.equals(this.value, resolved.value) && Objects.equals(this.exception, resolved.exception);
+	}
+
+	@Override
+	public int hashCode() {
+		return Objects.hash(this.value, this.exception);
+	}
+
+	@Override
+	public String toString() {
+		return new StringJoiner(", ", Resolved.class.getSimpleName() + "[", "]").add("value=" + this.value)
+			.add("exception=" + this.exception)
+			.toString();
+	}
+
 }

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

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2012-2024 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 static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.config.BeanExpressionContext;
+import org.springframework.beans.factory.config.BeanExpressionResolver;
+import org.springframework.beans.factory.config.ConfigurableBeanFactory;
+
+/**
+ * Unit tests for {@link DefaultExpressionResolver}.
+ *
+ * @author Jonas Geiregat
+ */
+class DefaultExpressionResolverTest {
+
+	@Test
+	void resolveEvaluatedStringResult() {
+		var configurableBeanFactory = mock(ConfigurableBeanFactory.class);
+		var beanExpressionResolver = mock(BeanExpressionResolver.class);
+		when(configurableBeanFactory.getBeanExpressionResolver()).thenReturn(beanExpressionResolver);
+		when(configurableBeanFactory.resolveEmbeddedValue("${topic.name}")).thenReturn("resolved-topic-name");
+		when(beanExpressionResolver.evaluate(eq("resolved-topic-name"), any(BeanExpressionContext.class)))
+			.thenReturn("resolved-topic-name");
+
+		ExpressionResolver expressionResolver = new DefaultExpressionResolver(configurableBeanFactory);
+		Resolved<String> resolved = expressionResolver.resolveToString("${topic.name}");
+
+		assertThat(resolved).isEqualTo(Resolved.of("resolved-topic-name"));
+	}
+
+	@Test
+	void resolveEvaluatedNullResult() {
+		var configurableBeanFactory = mock(ConfigurableBeanFactory.class);
+		var beanExpressionResolver = mock(BeanExpressionResolver.class);
+		when(configurableBeanFactory.getBeanExpressionResolver()).thenReturn(beanExpressionResolver);
+		when(configurableBeanFactory.resolveEmbeddedValue("#{null")).thenReturn("#{null}");
+		when(beanExpressionResolver.evaluate(eq("#{null}"), any(BeanExpressionContext.class))).thenReturn(null);
+
+		ExpressionResolver expressionResolver = new DefaultExpressionResolver(configurableBeanFactory);
+		Resolved<String> resolved = expressionResolver.resolveToString("#{null}");
+
+		assertThat(resolved).isEqualTo(Resolved.of(null));
+	}
+
+	@Test
+	void failToResolveEvaluatedNoneString() {
+		var configurableBeanFactory = mock(ConfigurableBeanFactory.class);
+		var beanExpressionResolver = mock(BeanExpressionResolver.class);
+		when(configurableBeanFactory.getBeanExpressionResolver()).thenReturn(beanExpressionResolver);
+		when(configurableBeanFactory.resolveEmbeddedValue("#{someBean.someProperty}"))
+			.thenReturn("#{someBean.someProperty}");
+		when(beanExpressionResolver.evaluate(eq("#{someBean.someProperty}"), any(BeanExpressionContext.class)))
+			.thenReturn(new Object() {
+			});
+
+		ExpressionResolver expressionResolver = new DefaultExpressionResolver(configurableBeanFactory);
+		Resolved<String> resolved = expressionResolver.resolveToString("#{someBean.someProperty}");
+
+		assertThat(resolved.exception())
+			.hasValueSatisfying((ex) -> assertThat(ex).isInstanceOf(IllegalArgumentException.class)
+				.hasMessageContaining("The expression '#{someBean.someProperty}' must resolve to a string but was: "));
+
+	}
+
+}