diff --git a/docs/src/main/asciidoc/sqs.adoc b/docs/src/main/asciidoc/sqs.adoc index 9b69146b0..1f4de984a 100644 --- a/docs/src/main/asciidoc/sqs.adoc +++ b/docs/src/main/asciidoc/sqs.adoc @@ -720,6 +720,15 @@ public class MyListener { The `isDefault = true` parameter designates a method as the fallback handler for messages that don't match any other handler's parameter type. +Handler methods can also accept the payload wrapped in a `Message`; routing is based on the generic type, so it works the same way as a plain payload parameter: +[source, java] +---- +@SqsHandler +public void handle(Message message) { + System.out.println(message.getPayload()); +} +---- + To determine which handler method to invoke, the framework needs to know the payload type before deserialization. A custom `PayloadTypeMapper` should be configured to map incoming messages to their concrete types. See <> for an example. diff --git a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/adapter/CompositeInvocableHandler.java b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/adapter/CompositeInvocableHandler.java index 0d602e70d..c14d4249a 100644 --- a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/adapter/CompositeInvocableHandler.java +++ b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/adapter/CompositeInvocableHandler.java @@ -22,6 +22,7 @@ import java.util.concurrent.ConcurrentMap; import org.jspecify.annotations.Nullable; import org.springframework.core.MethodParameter; +import org.springframework.core.ResolvableType; import org.springframework.messaging.Message; import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; @@ -173,13 +174,20 @@ private MethodParameter findCandidate(Class payloadClass, Method method) { } /** - * Checks if the given method parameter is assignable from the payload type. + * Checks if the given method parameter is assignable from the payload type. For a {@link Message} parameter, the + * payload is matched against the resolved generic type. * * @param methodParameter the method parameter to check * @param payloadClass the class of the payload * @return true if the parameter type is assignable from the payload type, false otherwise */ private boolean isPayloadAssignable(MethodParameter methodParameter, Class payloadClass) { - return methodParameter.getParameterType().isAssignableFrom(payloadClass); + Class parameterType = methodParameter.getParameterType(); + if (Message.class.isAssignableFrom(parameterType)) { + ResolvableType resolvableType = ResolvableType.forMethodParameter(methodParameter); + Class genericType = resolvableType.getGeneric(0).resolve(); + return genericType != null && genericType.isAssignableFrom(payloadClass); + } + return parameterType.isAssignableFrom(payloadClass); } } diff --git a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsIntegrationTests.java b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsIntegrationTests.java index 46e1f7b46..c28cadaa6 100644 --- a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsIntegrationTests.java +++ b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsIntegrationTests.java @@ -121,6 +121,8 @@ class SqsIntegrationTests extends BaseSqsIntegrationTest { static final String RECEIVES_MESSAGE_MULTI_METHOD_QUEUE_NAME = "receives_message_multi_method_test_queue"; + static final String RECEIVES_MESSAGE_MULTI_METHOD_WITH_MESSAGE_ENVELOPE_QUEUE_NAME = "receives_message_multi_method_with_message_envelope_test_queue"; + static final String RECEIVES_MESSAGE_ASYNC_QUEUE_NAME = "receives_message_async_test_queue"; static final String DOES_NOT_ACK_ON_ERROR_QUEUE_NAME = "does_not_ack_test_queue"; @@ -223,6 +225,18 @@ void receivesMessageOnMultiMethod() throws Exception { assertThat(latchContainer.receivesMessageMultiMethodLatch.await(10, TimeUnit.SECONDS)).isTrue(); } + @Test + void receivesMessageOnMultiMethodWithMessageEnvelope() throws Exception { + String payload = "receivesMessageOnMultiMethodWithMessageEnvelope-payload"; + + sqsTemplate.send(RECEIVES_MESSAGE_MULTI_METHOD_WITH_MESSAGE_ENVELOPE_QUEUE_NAME, payload); + logger.debug("Sent message to queue {} with messageBody {}", + RECEIVES_MESSAGE_MULTI_METHOD_WITH_MESSAGE_ENVELOPE_QUEUE_NAME, payload); + + assertThat(latchContainer.receivesMessageMultiMethodMessageEnvelopeLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(latchContainer.receivesMessageMultiMethodDefaultHandlerLatch.getCount()).isEqualTo(1); + } + @Test void observesMessage() throws Exception { String messageBody = "observesMessage-payload"; @@ -533,6 +547,25 @@ void handle(Object message) { } } + @SqsListener(queueNames = RECEIVES_MESSAGE_MULTI_METHOD_WITH_MESSAGE_ENVELOPE_QUEUE_NAME, pollTimeoutSeconds = "${property.one}", maxMessagesPerPoll = "${property.one}", maxConcurrentMessages = "${missing.property:5}", id = "receivesMessageMultiMethodWithMessageEnvelopeListener") + static class ReceivesMessageMultiMethodWithMessageEnvelopeListener { + + @Autowired + LatchContainer latchContainer; + + @SqsHandler + void handle(Message message) { + logger.debug("Received Message in Listener Method: " + message.getPayload()); + latchContainer.receivesMessageMultiMethodMessageEnvelopeLatch.countDown(); + } + + @SqsHandler(isDefault = true) + void handleDefault(Object message) { + logger.debug("Received default message in Listener Method: " + message); + latchContainer.receivesMessageMultiMethodDefaultHandlerLatch.countDown(); + } + } + static class ReceivesMessageBatchListener { @Autowired @@ -711,6 +744,8 @@ static class LatchContainer { final CountDownLatch receivesMessageBatchLatch = new CountDownLatch(20); final CountDownLatch receivesMessageAsyncLatch = new CountDownLatch(1); final CountDownLatch receivesMessageMultiMethodLatch = new CountDownLatch(3); + final CountDownLatch receivesMessageMultiMethodMessageEnvelopeLatch = new CountDownLatch(1); + final CountDownLatch receivesMessageMultiMethodDefaultHandlerLatch = new CountDownLatch(1); final CountDownLatch doesNotAckLatch = new CountDownLatch(2); final CountDownLatch doesNotAckAsyncLatch = new CountDownLatch(2); final CountDownLatch doesNotAckBatchLatch = new CountDownLatch(20); @@ -945,6 +980,11 @@ ReceivesMessageMultiMethodListener receivesMessageMultiMethodListener() { return new ReceivesMessageMultiMethodListener(); } + @Bean + ReceivesMessageMultiMethodWithMessageEnvelopeListener receivesMessageMultiMethodWithMessageEnvelopeListener() { + return new ReceivesMessageMultiMethodWithMessageEnvelopeListener(); + } + @Bean DoesNotAckOnErrorListener doesNotAckOnErrorListener() { return new DoesNotAckOnErrorListener(); diff --git a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/adapter/CompositeInvocableHandlerTest.java b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/adapter/CompositeInvocableHandlerTest.java index 58751e10d..9b68634c9 100644 --- a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/adapter/CompositeInvocableHandlerTest.java +++ b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/adapter/CompositeInvocableHandlerTest.java @@ -49,6 +49,18 @@ public String handle(Integer payload) { public String handle(Object payload) { return "Handled object: " + payload; } + + public String handle(Message message) { + return "Handled message: " + message.getPayload(); + } + + public String handleRaw(Message message) { + return "Handled raw: " + message.getPayload(); + } + + public String handlePayloadAndMessage(String payload, Message message) { + return "Handled payload and message: " + payload; + } } @Test @@ -106,8 +118,69 @@ void throwsIfMultipleHandlersMatch() throws Exception { .hasMessageContaining("Ambiguous handler method for payload type"); } + @Test + void invokesHandlerWithMessageParameterMatchingPayloadType() throws Exception { + Message message = mock(Message.class); + when(message.getPayload()).thenReturn("testPayload"); + + InvocableHandlerMethod messageHandler = mockHandler(Message.class, "messageHandlerResult"); + InvocableHandlerMethod intHandler = mockHandler(Integer.class, "shouldNotBeCalled"); + + CompositeInvocableHandler handler = new CompositeInvocableHandler(List.of(messageHandler, intHandler), null); + + assertThat(handler.invoke(message)).isEqualTo("messageHandlerResult"); + verify(messageHandler).invoke(message); + verify(intHandler, never()).invoke(message); + } + + @Test + void throwsWhenPlainAndMessageHandlersMatchSamePayloadType() throws Exception { + Message message = mock(Message.class); + when(message.getPayload()).thenReturn("testPayload"); + + InvocableHandlerMethod stringHandler = mockHandler(String.class, "stringHandlerResult"); + InvocableHandlerMethod messageHandler = mockHandler(Message.class, "messageHandlerResult"); + + CompositeInvocableHandler handler = new CompositeInvocableHandler(List.of(stringHandler, messageHandler), null); + + assertThatThrownBy(() -> handler.invoke(message)).isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Ambiguous handler method for payload type"); + } + + @Test + void rawMessageParameterDoesNotMatchAndFallsBackToDefault() throws Exception { + Message message = mock(Message.class); + when(message.getPayload()).thenReturn("testPayload"); + + InvocableHandlerMethod rawHandler = mockHandler("handleRaw", "shouldNotBeCalled", Message.class); + InvocableHandlerMethod defaultHandler = mockHandler(Object.class, "defaultResult"); + + CompositeInvocableHandler composite = new CompositeInvocableHandler(List.of(rawHandler), defaultHandler); + + assertThat(composite.invoke(message)).isEqualTo("defaultResult"); + } + + @Test + void throwsWhenPayloadAndMessageParametersOnSameMethod() throws Exception { + Message message = mock(Message.class); + when(message.getPayload()).thenReturn("testPayload"); + + InvocableHandlerMethod handler = mockHandler("handlePayloadAndMessage", "result", String.class, Message.class); + InvocableHandlerMethod intHandler = mockHandler(Integer.class, "shouldNotBeCalled"); + + CompositeInvocableHandler composite = new CompositeInvocableHandler(List.of(handler, intHandler), null); + + assertThatThrownBy(() -> composite.invoke(message)).isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Ambiguous payload parameter for"); + } + private InvocableHandlerMethod mockHandler(Class paramType, Object returnValue) throws Exception { - Method method = Listener.class.getDeclaredMethod("handle", paramType); + return mockHandler("handle", returnValue, paramType); + } + + private InvocableHandlerMethod mockHandler(String methodName, Object returnValue, Class... paramTypes) + throws Exception { + Method method = Listener.class.getDeclaredMethod(methodName, paramTypes); InvocableHandlerMethod handler = mock(InvocableHandlerMethod.class); when(handler.getMethod()).thenReturn(method); when(handler.invoke(any())).thenReturn(returnValue);