Add MessageUtils for Reactive message handling (#510)

Eases the burden of converting from a Spring message to a Pulsar
MessageId in order to pass into the Pulsar MessageResult ack/nack
methods.

Resolves #509

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

@@ -213,15 +213,16 @@ The following example uses `ReactivePulsarListener` to consume a stream of POJOs
 [source, java]
 ----
 @ReactivePulsarListener(topics = "streaming-1", stream = true)
-Flux<MessageResult<Void>> listen(Flux<Message<String>> messages) {
+Flux<MessageResult<Void>> listen(Flux<org.apache.pulsar.client.api.Message<String>> messages) {
     return messages
         .doOnNext((msg) -> System.out.println("Received: " + msg.getValue()))
         .map(MessageResult::acknowledge);
 ----
-Here we receive the records as a `Flux` of messages.
+Here we receive the records as a `Flux` of Pulsar messages.
 In addition, to enable stream consumption at the `ReactivePulsarListener` level, you need to set the `stream` property on the annotation to `true`.
 
-NOTE: The listener method returns a `Flux<MessageResult<Void>>` where each element represents a processed message and holds the message id, value and whether it was acknowledged. The `MessageResult` has a set of static factory methods that can be used to create the appropriate `MessageResult` instance.
+NOTE: The listener method returns a `Flux<MessageResult<Void>>` where each element represents a processed message and holds the message id, value and whether it was acknowledged.
+The `MessageResult` has a set of static factory methods that can be used to create the appropriate `MessageResult` instance.
 
 Based on the actual type of the messages in the `Flux`, the framework tries to infer the schema to use.
 If it contains a complex type, you still need to provide the `schemaType` on `ReactivePulsarListener`.
@@ -233,10 +234,13 @@ The following listener uses the Spring messaging `Message` envelope with a compl
 Flux<MessageResult<Void>> listen2(Flux<org.springframework.messaging.Message<Foo>> messages) {
     return messages
         .doOnNext((msg) -> System.out.println("Received: " + msg.getPayload()))
-        .map(MessageResult::acknowledge);
+        .map(MessageUtils::acknowledge);
 }
 ----
 
+NOTE: The listener method returns a `Flux<MessageResult<Void>>` where each element represents a processed message and holds the message id, value and whether it was acknowledged.
+The Spring `MessageUtils` has a set of static factory methods that can be used to create the appropriate `MessageResult` instance from a Spring message.
+
 ==== Configuration - Application Properties
 The listener relies on the `ReactivePulsarConsumerFactory` to create and manage the underlying Pulsar consumer that it uses to consume messages.
 Spring Boot provides this consumer factory which you can further configure by specifying the {spring-boot-pulsar-config-props}[`spring.pulsar.consumer.*`] application properties.

spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/support/MessageUtils.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2023-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.reactive.support;
+
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.reactive.client.api.MessageResult;
+
+import org.springframework.messaging.Message;
+import org.springframework.pulsar.support.PulsarHeaders;
+
+/**
+ * Convenience functions related to Spring {@link Message messages}.
+ *
+ * @author Chris Bono
+ */
+public final class MessageUtils {
+
+	private MessageUtils() {
+	}
+
+	/**
+	 * Determine the Pulsar {@link MessageId} for a given Spring message by extracting the
+	 * value of its {@link PulsarHeaders#MESSAGE_ID} header.
+	 * @param <T> the type of message payload
+	 * @param message the Spring message
+	 * @return the Pulsar message id
+	 * @throws IllegalStateException if the message id could not be determined
+	 */
+	public static <T> MessageId extractMessageId(Message<T> message) {
+		if (message.getHeaders().get(PulsarHeaders.MESSAGE_ID) instanceof MessageId msgId) {
+			return msgId;
+		}
+		throw new IllegalStateException("Spring Message missing '%s' header".formatted(PulsarHeaders.MESSAGE_ID));
+	}
+
+	/**
+	 * Convenience method that acknowledges a Spring message by {@link #extractMessageId
+	 * extracting} its message id and passing it to
+	 * {@link MessageResult#acknowledge(MessageId)}.
+	 * @param <T> the type of message payload
+	 * @param message the Spring message to acknowledge
+	 * @return an empty value and signals that the message must be acknowledged
+	 */
+	public static <T> MessageResult<Void> acknowledge(Message<T> message) {
+		return MessageResult.acknowledge(MessageUtils.extractMessageId(message));
+	}
+
+	/**
+	 * Convenience method that negatively acknowledges a Spring message by
+	 * {@link #extractMessageId extracting} its message id and passing it to
+	 * {@link MessageResult#negativeAcknowledge(MessageId)}.
+	 * @param <T> the type of message payload
+	 * @param message the Spring message to negatively acknowledge
+	 * @return an empty value and signals that the message must be negatively acknowledged
+	 */
+	public static <T> MessageResult<Void> negativeAcknowledge(Message<T> message) {
+		return MessageResult.negativeAcknowledge(MessageUtils.extractMessageId(message));
+	}
+
+}

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

@@ -74,6 +74,7 @@ import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTests.
 import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTests.StreamingListenerTestCases.StreamingListenerTestCasesConfig;
 import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTests.SubscriptionTypeTests.WithDefaultType.WithDefaultTypeConfig;
 import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTests.SubscriptionTypeTests.WithSpecificTypes.WithSpecificTypesConfig;
+import org.springframework.pulsar.reactive.support.MessageUtils;
 import org.springframework.pulsar.support.PulsarHeaders;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.util.ReflectionTestUtils;
@@ -215,15 +216,7 @@ class ReactivePulsarListenerTests extends ReactivePulsarListenerTestsBase {
 			@ReactivePulsarListener(topics = "streaming-2", stream = true,
 					consumerCustomizer = "subscriptionInitialPositionEarliest")
 			Flux<MessageResult<Void>> listen2(Flux<org.springframework.messaging.Message<String>> messages) {
-				return messages.doOnNext(m -> latch2.countDown()).map(m -> {
-					Object mId = m.getHeaders().get(PulsarHeaders.MESSAGE_ID);
-					if (mId instanceof MessageId) {
-						return (MessageId) mId;
-					}
-					else {
-						throw new RuntimeException("Missing message Id");
-					}
-				}).map(MessageResult::acknowledge);
+				return messages.doOnNext(m -> latch2.countDown()).map(MessageUtils::acknowledge);
 			}
 
 		}

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

@@ -25,7 +25,6 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 
-import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.common.schema.SchemaType;
@@ -46,6 +45,7 @@ import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTombst
 import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTombstoneTests.SpringMessagePayload.SpringMessagePayloadConfig;
 import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTombstoneTests.StreamingPulsarMessagePayload.StreamingPulsarMessagePayloadConfig;
 import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTombstoneTests.StreamingSpringMessagePayload.StreamingSpringMessagePayloadConfig;
+import org.springframework.pulsar.reactive.support.MessageUtils;
 import org.springframework.pulsar.support.PulsarHeaders;
 import org.springframework.pulsar.support.PulsarNull;
 import org.springframework.test.context.ContextConfiguration;
@@ -336,14 +336,7 @@ class ReactivePulsarListenerTombstoneTests extends ReactivePulsarListenerTestsBa
 					var keyHeader = (String) m.getHeaders().get(PulsarHeaders.KEY);
 					receivedMessagesWithHeaders.add(new ReceivedMessage<>(payload, keyHeader));
 					latchWithHeaders.countDown();
-				}).map(m -> this.messageIdFrom(m)).map(MessageResult::acknowledge);
-			}
-
-			private <T> MessageId messageIdFrom(Message<T> springMessage) {
-				if (springMessage.getHeaders().get(PulsarHeaders.MESSAGE_ID) instanceof MessageId msgId) {
-					return msgId;
-				}
-				throw new RuntimeException("Spring Message missing '%s' header".formatted(PulsarHeaders.MESSAGE_ID));
+				}).map(MessageUtils::acknowledge);
 			}
 
 		}

spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/support/MessageUtilsTests.java

@@ -0,0 +1,102 @@
+/*
+ * Copyright 2023-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.reactive.support;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
+
+import java.util.Map;
+
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.reactive.client.api.MessageResult;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.pulsar.support.PulsarHeaders;
+
+/**
+ * Tests for {@link MessageUtils}.
+ *
+ * @author Chris Bono
+ */
+class MessageUtilsTests {
+
+	@Nested
+	class ExtractMessageIdApi {
+
+		@Test
+		void shouldReturnMessageIdWhenValidHeader() {
+			var msgId = mock(MessageId.class);
+			var msg = new GenericMessage<>("m1", Map.of(PulsarHeaders.MESSAGE_ID, msgId));
+			assertThat(MessageUtils.extractMessageId(msg)).isEqualTo(msgId);
+		}
+
+		@Test
+		void shouldThrowExceptionWhenInvalidHeader() {
+			var msg = new GenericMessage<>("m1", Map.of(PulsarHeaders.MESSAGE_ID, "badId"));
+			assertThatIllegalStateException().isThrownBy(() -> MessageUtils.extractMessageId(msg))
+				.withMessage("Spring Message missing 'pulsar_message_id' header");
+		}
+
+		@Test
+		void shouldThrowExceptionWhenEmptyHeaders() {
+			var msg = new GenericMessage<>("m1");
+			assertThatIllegalStateException().isThrownBy(() -> MessageUtils.extractMessageId(msg))
+				.withMessage("Spring Message missing 'pulsar_message_id' header");
+		}
+
+	}
+
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@Nested
+	class AcknowledgeApi {
+
+		@Test
+		void shouldDelegateToMessageResultAcknowledge() {
+			var msgId = mock(MessageId.class);
+			var msg = new GenericMessage<>("m1", Map.of(PulsarHeaders.MESSAGE_ID, msgId));
+			try (MockedStatic<MessageResult> messageResult = mockStatic(MessageResult.class)) {
+				var mockedReturnValue = (MessageResult<Void>) mock(MessageResult.class);
+				when(MessageResult.acknowledge(any(MessageId.class))).thenReturn(mockedReturnValue);
+				var returnedResult = MessageUtils.acknowledge(msg);
+				assertThat(returnedResult).isEqualTo(mockedReturnValue);
+				messageResult.verify(() -> MessageResult.acknowledge(msgId));
+			}
+		}
+
+		@Test
+		void shouldDelegateToMessageResultNegativeAcknowledge() {
+			var msgId = mock(MessageId.class);
+			var msg = new GenericMessage<>("m1", Map.of(PulsarHeaders.MESSAGE_ID, msgId));
+			try (MockedStatic<MessageResult> messageResult = mockStatic(MessageResult.class)) {
+				var mockedReturnValue = (MessageResult<Void>) mock(MessageResult.class);
+				when(MessageResult.negativeAcknowledge(any(MessageId.class))).thenReturn(mockedReturnValue);
+				var returnedResult = MessageUtils.negativeAcknowledge(msg);
+				assertThat(returnedResult).isEqualTo(mockedReturnValue);
+				messageResult.verify(() -> MessageResult.negativeAcknowledge(msgId));
+			}
+		}
+
+	}
+
+}