Add support for multi-method listeners

Author: joseiedo

docs/src/main/asciidoc/sqs.adoc

@@ -632,6 +632,34 @@ Any number of `@SqsListener` annotations can be used in a bean class, and each a
 
 NOTE: Queues declared in the same annotation will share the container, though each will have separate throughput and acknowledgement controls.
 
+===== Handling Different Payload In The Same Listener
+
+It's possible to handle different payloads in the same listener by annotating handler methods with the `@SqsHandler` annotation.
+Here's a sample:
+[source, java]
+----
+@SqsListener("myQueue")
+public class MyListener {
+
+	@SqsHandler
+	public void handle(String message) {
+		System.out.println(message);
+	}
+
+	@SqsHandler
+	public void handle(MyPojo pojo) {
+		System.out.println(pojo);
+	}
+
+	@SqsHandler(isDefault = true)
+	public void handle(Object message) {
+		System.out.println(message);
+	}
+}
+----
+
+The `isDefault = true` parameter designates a method as the fallback handler for messages that don't match any other handler's parameter type.
+
 ===== SNS Messages
 
 Since 3.1.1, when receiving SNS messages through the `@SqsListener`, the message includes all attributes of the `SnsNotification`. To only receive need the `Message` part of the payload, you can utilize the `@SnsNotificationMessage` annotation.

spring-cloud-aws-samples/spring-cloud-aws-sqs-sample/src/main/java/io/awspring/cloud/sqs/sample/SpringSqsHandlerSample.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2013-2021 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 io.awspring.cloud.sqs.sample;
+
+import io.awspring.cloud.sqs.annotation.SqsHandler;
+import io.awspring.cloud.sqs.annotation.SqsListener;
+import io.awspring.cloud.sqs.operations.SqsTemplate;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * Sample class to demonstrate how to handle multiple message types in a single listener
+ * with {@link SqsHandler} annotation.
+ *
+ * @author José Iêdo
+ */
+@Configuration
+@SqsListener(queueNames = SpringSqsHandlerSample.QUEUE_NAME)
+public class SpringSqsHandlerSample {
+
+	public static final String QUEUE_NAME = "multi-method-queue";
+	private static final Logger LOGGER = LoggerFactory.getLogger(SpringSqsHandlerSample.class);
+
+	private interface BaseMessage { }
+	private record SampleRecord(String propertyOne, String propertyTwo) { }
+	private record AnotherSampleRecord(String propertyOne, String propertyTwo) implements BaseMessage { }
+
+	@SqsHandler
+	void handleMessage(SampleRecord message) {
+		LOGGER.info("Received message of type SampleRecord: {}", message);
+	}
+
+	@SqsHandler
+	void handleMessage(BaseMessage message) {
+		LOGGER.info("Received message of type BaseMessage: {}", message);
+	}
+
+	@SqsHandler(isDefault = true)
+	void handleMessage(Object message) {
+		LOGGER.info("Received message of type Object: {}", message);
+	}
+
+
+	@Bean
+	public ApplicationRunner sendMessageToQueueWithMultipleHandlers(SqsTemplate sqsTemplate) {
+		return args -> {
+			sqsTemplate.send(QUEUE_NAME, new SampleRecord("Hello!", "From SQS!"));
+			sqsTemplate.send(QUEUE_NAME, new AnotherSampleRecord("Hello!", "From SQS!"));
+			sqsTemplate.send(QUEUE_NAME, "Hello!");
+		};
+	}
+}

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/annotation/AbstractListenerAnnotationBeanPostProcessor.java

@@ -35,6 +35,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -76,6 +77,7 @@
  *
  * @author Tomaz Fernandes
  * @author Joao Calassio
+ * @author José Iêdo
  * @since 3.0
  */
 public abstract class AbstractListenerAnnotationBeanPostProcessor<A extends Annotation>
@@ -120,16 +122,33 @@ protected void detectAnnotationsAndRegisterEndpoints(Object bean, Class<?> targe
 		Map<Method, A> annotatedMethods = MethodIntrospector.selectMethods(targetClass,
 				(MethodIntrospector.MetadataLookup<A>) method -> AnnotatedElementUtils.findMergedAnnotation(method,
 						getAnnotationClass()));
-		if (annotatedMethods.isEmpty()) {
+
+		A classListener = AnnotatedElementUtils.findMergedAnnotation(targetClass, getAnnotationClass());
+		boolean hasMethodLevelListeners = !annotatedMethods.isEmpty();
+		boolean hasClassLevelListeners = classListener != null;
+
+		if (!hasMethodLevelListeners && !hasClassLevelListeners) {
 			this.nonAnnotatedClasses.add(targetClass);
 		}
-		annotatedMethods.entrySet().stream()
-				.map(entry -> createAndConfigureEndpoint(bean, entry.getKey(), entry.getValue()))
-				.forEach(this.endpointRegistrar::registerEndpoint);
+		else {
+			if (hasMethodLevelListeners) {
+				annotatedMethods.entrySet().stream()
+						.map(entry -> createAndConfigureEndpoint(bean, entry.getKey(), entry.getValue()))
+						.forEach(this.endpointRegistrar::registerEndpoint);
+			}
+
+			if (hasClassLevelListeners) {
+				Set<Method> handlerMethods = getHandlerMethods(targetClass);
+				createAndConfigureMultiMethodEndpoint(bean, targetClass, classListener,
+						new ArrayList<>(handlerMethods));
+			}
+		}
 	}
 
 	protected abstract Class<A> getAnnotationClass();
 
+	protected abstract Set<Method> getHandlerMethods(Class<?> targetClass);
+
 	private Endpoint createAndConfigureEndpoint(Object bean, Method method, A annotation) {
 		Endpoint endpoint = createEndpoint(annotation);
 		ConfigUtils.INSTANCE.acceptIfInstance(endpoint, HandlerMethodEndpoint.class, hme -> {
@@ -140,8 +159,30 @@ private Endpoint createAndConfigureEndpoint(Object bean, Method method, A annota
 		return endpoint;
 	}
 
+	private void createAndConfigureMultiMethodEndpoint(Object bean, Class<?> targetClass, A classListener,
+			List<Method> handlerMethods) {
+		Assert.notEmpty(handlerMethods, "No handler method found for listener in class: " + targetClass);
+		Method defaultMethod = getDefaultHandlerMethod(targetClass, handlerMethods);
+
+		Endpoint endpoint = createMultiMethodEndpoint(classListener, handlerMethods, defaultMethod, bean);
+		for (Method method : handlerMethods) {
+			ConfigUtils.INSTANCE.acceptIfInstance(endpoint, HandlerMethodEndpoint.class, hme -> {
+				hme.setBean(bean);
+				hme.setMethod(method);
+				hme.setHandlerMethodFactory(this.delegatingHandlerMethodFactory);
+			});
+		}
+
+		this.endpointRegistrar.registerEndpoint(endpoint);
+	}
+
+	protected abstract Method getDefaultHandlerMethod(Class<?> targetClass, List<Method> handlerMethods);
+
 	protected abstract Endpoint createEndpoint(A sqsListenerAnnotation);
 
+	protected abstract Endpoint createMultiMethodEndpoint(A sqsListenerAnnotation, List<Method> methods,
+			@Nullable Method defaultMethod, Object bean);
+
 	protected Collection<String> resolveEndpointNames(String[] endpointNames) {
 		return Arrays.stream(endpointNames).map(this::resolveExpression)
 				.flatMap(resolvedName -> resolveAsStrings(resolvedName).stream()).collect(Collectors.toList());

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/annotation/SqsHandler.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2013-2025 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 io.awspring.cloud.sqs.annotation;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+import org.springframework.messaging.handler.annotation.MessageMapping;
+
+/**
+ * Methods that are from classes annotated with {@link SqsListener} and are annotated with {@link SqsHandler} will be
+ * marked as the target of the SQS message listener based on the message payload type.
+ *
+ * <p>
+ * Each payload type must have exactly one corresponding method.
+ * <p>
+ * If no method matches the payload type, a method marked as the default (using {@code isDefault = true}) will be invoked.
+ * Only one method can be designated as the default.
+ *
+ * @author José Iêdo
+ */
+@Target(ElementType.METHOD)
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@MessageMapping
+public @interface SqsHandler {
+
+	/**
+	 * Indicates whether this method should be used as the default fallback method if no other {@link SqsHandler}
+	 * method matches the payload type.
+	 *
+	 * @return {@code true} if this is the default method, {@code false} otherwise
+	 */
+	boolean isDefault() default false;
+}

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/annotation/SqsListener.java

@@ -76,9 +76,10 @@
  * @author Matej Nedic
  * @author Tomaz Fernandes
  * @author Joao Calassio
+ * @author José Iêdo
  * @since 1.1
  */
-@Target(ElementType.METHOD)
+@Target({ ElementType.METHOD, ElementType.TYPE })
 @Retention(RetentionPolicy.RUNTIME)
 @Documented
 @MessageMapping

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/annotation/SqsListenerAnnotationBeanPostProcessor.java

@@ -17,6 +17,7 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.awspring.cloud.sqs.config.Endpoint;
+import io.awspring.cloud.sqs.config.MultiMethodSqsEndpoint;
 import io.awspring.cloud.sqs.config.SqsBeanNames;
 import io.awspring.cloud.sqs.config.SqsEndpoint;
 import io.awspring.cloud.sqs.listener.SqsHeaders;
@@ -26,18 +27,25 @@
 import io.awspring.cloud.sqs.support.resolver.QueueAttributesMethodArgumentResolver;
 import io.awspring.cloud.sqs.support.resolver.SqsMessageMethodArgumentResolver;
 import io.awspring.cloud.sqs.support.resolver.VisibilityHandlerMethodArgumentResolver;
+import java.lang.reflect.Method;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
+import java.util.Set;
+import org.springframework.core.MethodIntrospector;
+import org.springframework.core.annotation.AnnotationUtils;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver;
+import org.springframework.util.ReflectionUtils;
 
 /**
  * {@link AbstractListenerAnnotationBeanPostProcessor} implementation for {@link SqsListener @SqsListener}.
  *
  * @author Tomaz Fernandes
  * @author Joao Calassio
+ * @author José Iêdo
  * @since 3.0
  */
 public class SqsListenerAnnotationBeanPostProcessor extends AbstractListenerAnnotationBeanPostProcessor<SqsListener> {
@@ -62,6 +70,22 @@ protected Endpoint createEndpoint(SqsListener sqsListenerAnnotation) {
 				.acknowledgementMode(resolveAcknowledgement(sqsListenerAnnotation.acknowledgementMode())).build();
 	}
 
+	@Override
+	protected Endpoint createMultiMethodEndpoint(SqsListener sqsListenerAnnotation, List<Method> methods,
+			@Nullable Method defaultMethod, Object bean) {
+		return MultiMethodSqsEndpoint.builder()
+				.factoryBeanName(resolveAsString(sqsListenerAnnotation.factory(), "factory"))
+				.queueNames(resolveEndpointNames(sqsListenerAnnotation.value())).bean(bean).methods(methods)
+				.defaultMethod(defaultMethod).id(getEndpointId(sqsListenerAnnotation.id()))
+				.sqsEndpoint(createEndpoint(sqsListenerAnnotation)).build();
+	}
+
+	@Override
+	protected Set<Method> getHandlerMethods(Class<?> targetClass) {
+		return MethodIntrospector.selectMethods(targetClass, (ReflectionUtils.MethodFilter) method -> AnnotationUtils
+				.findAnnotation(method, SqsHandler.class) != null);
+	}
+
 	@Override
 	protected String getGeneratedIdPrefix() {
 		return GENERATED_ID_PREFIX;
@@ -90,4 +114,21 @@ protected Collection<HandlerMethodArgumentResolver> createAdditionalArgumentReso
 		return argumentResolvers;
 	}
 
+	@Override
+	protected Method getDefaultHandlerMethod(Class<?> targetClass, List<Method> handlerMethods) {
+		Method defaultMethod = null;
+		for (Method method : handlerMethods) {
+			SqsHandler annotation = method.getAnnotation(SqsHandler.class);
+			if (annotation.isDefault()) {
+				if (defaultMethod != null) {
+					throw new IllegalArgumentException(
+							"There is more than one default method for the same listener in class: " + targetClass);
+				}
+				else {
+					defaultMethod = method;
+				}
+			}
+		}
+		return defaultMethod;
+	}
 }

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/config/AbstractEndpoint.java

@@ -42,6 +42,7 @@
  * Base class for implementing a {@link HandlerMethodEndpoint}.
  *
  * @author Tomaz Fernandes
+ * @author José Iêdo
  * @since 3.0
  */
 public abstract class AbstractEndpoint implements HandlerMethodEndpoint {
@@ -90,6 +91,14 @@ public void setBean(Object bean) {
 		this.bean = bean;
 	}
 
+	/**
+	 * Get the bean instance to be used when handling a message for this endpoint.
+	 * @return the bean instance.
+	 */
+	public Object getBean() {
+		return this.bean;
+	}
+
 	/**
 	 * Set the method to be used when handling a message for this endpoint.
 	 * @param method the method.
@@ -109,6 +118,15 @@ public void setHandlerMethodFactory(MessageHandlerMethodFactory handlerMethodFac
 		this.handlerMethodFactory = handlerMethodFactory;
 	}
 
+	/**
+	 * Get the {@link MessageHandlerMethodFactory} to be used for handling messages in this endpoint.
+	 * @return the factory.
+	 */
+	@Nullable
+	public MessageHandlerMethodFactory getMessageHandlerMethodFactory() {
+		return this.handlerMethodFactory;
+	}
+
 	@Override
 	public void configureListenerMode(Consumer<ListenerMode> consumer) {
 		List<MethodParameter> parameters = getMethodParameters();