From 0f2ae02e62706c651b7a6e7bb49ee6cae47d85dd Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Sun, 10 Dec 2023 12:34:15 -0600 Subject: [PATCH] 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 --- .../ROOT/pages/reference/reactive-pulsar.adoc | 12 ++- .../pulsar/reactive/support/MessageUtils.java | 74 +++++++++++++ .../listener/ReactivePulsarListenerTests.java | 11 +- .../ReactivePulsarListenerTombstoneTests.java | 11 +- .../reactive/support/MessageUtilsTests.java | 102 ++++++++++++++++++ 5 files changed, 188 insertions(+), 22 deletions(-) create mode 100644 spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/support/MessageUtils.java create mode 100644 spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/support/MessageUtilsTests.java diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/reactive-pulsar.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/reactive-pulsar.adoc index 31820bdd..01a6c3b3 100644 --- a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/reactive-pulsar.adoc +++ b/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> listen(Flux> messages) { +Flux> listen(Flux> 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>` 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>` 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> listen2(Flux> messages) { return messages .doOnNext((msg) -> System.out.println("Received: " + msg.getPayload())) - .map(MessageResult::acknowledge); + .map(MessageUtils::acknowledge); } ---- +NOTE: The listener method returns a `Flux>` 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. diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/support/MessageUtils.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/support/MessageUtils.java new file mode 100644 index 00000000..c8c39f56 --- /dev/null +++ b/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 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 MessageId extractMessageId(Message 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 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 MessageResult acknowledge(Message 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 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 MessageResult negativeAcknowledge(Message message) { + return MessageResult.negativeAcknowledge(MessageUtils.extractMessageId(message)); + } + +} diff --git a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java index bb8a6e14..05e8c124 100644 --- a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java +++ b/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> listen2(Flux> 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); } } diff --git a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTombstoneTests.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTombstoneTests.java index 291eefc5..5a1a6194 100644 --- a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTombstoneTests.java +++ b/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 MessageId messageIdFrom(Message 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); } } diff --git a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/support/MessageUtilsTests.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/support/MessageUtilsTests.java new file mode 100644 index 00000000..7333f1a2 --- /dev/null +++ b/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 = mockStatic(MessageResult.class)) { + var mockedReturnValue = (MessageResult) 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 = mockStatic(MessageResult.class)) { + var mockedReturnValue = (MessageResult) 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)); + } + } + + } + +}