Batch SnsNotification Support (#1191)

* Add SQS module @SnsNotificationMessage batch messages support

* Polishing

Fixes #1129

---------

Co-authored-by: maxjiang153 <[email protected]>

Author: tomazfernandes

docs/src/main/asciidoc/sqs.adoc

@@ -618,6 +618,9 @@ NOTE: Queues declared in the same annotation will share the container, though ea
 ===== 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.
+
+For handling individual message processing, the @SnsNotificationMessage annotation can be used in the following manner:
+
 [source, java]
 ----
 @SqsListener("my-queue")
@@ -626,6 +629,16 @@ public void listen(@SnsNotificationMessage Pojo pojo) {
 }
 ----
 
+For batch message processing, use the @SnsNotificationMessage annotation with a List<Pojo> parameter.
+
+[source, java]
+----
+@SqsListener("my-queue")
+public void listen(@SnsNotificationMessage List<Pojo> pojos) {
+	System.out.println(pojos.size());
+}
+----
+
 ===== Specifying a MessageListenerContainerFactory
 A `MessageListenerContainerFactory` can be specified through the `factory` property.
 Such factory will then be used to create the container for the annotated method.

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/support/converter/SnsMessageConverter.java

@@ -17,6 +17,12 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.util.List;
+import org.springframework.core.GenericTypeResolver;
+import org.springframework.core.MethodParameter;
+import org.springframework.core.ResolvableType;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
@@ -29,6 +35,7 @@
 /**
  * @author Michael Sosa
  * @author gustavomonarin
+ * @author Wei Jiang
  * @since 3.1.1
  */
 public class SnsMessageConverter implements SmartMessageConverter {
@@ -45,10 +52,31 @@ public SnsMessageConverter(MessageConverter payloadConverter, ObjectMapper jsonM
 	}
 
 	@Override
+	@SuppressWarnings("unchecked")
 	public Object fromMessage(Message<?> message, Class<?> targetClass, @Nullable Object conversionHint) {
 		Assert.notNull(message, "message must not be null");
 		Assert.notNull(targetClass, "target class must not be null");
 
+		Object payload = message.getPayload();
+
+		if (payload instanceof List messages) {
+			return fromGenericMessages(messages, targetClass, conversionHint);
+		}
+		else {
+			return fromGenericMessage((GenericMessage<?>) message, targetClass, conversionHint);
+		}
+	}
+
+	private Object fromGenericMessages(List<GenericMessage<?>> messages, Class<?> targetClass,
+			@Nullable Object conversionHint) {
+		Type resolvedType = getResolvedType(targetClass, conversionHint);
+		Class<?> resolvedClazz = ResolvableType.forType(resolvedType).resolve();
+
+		return messages.stream().map(message -> fromGenericMessage(message, resolvedClazz, resolvedType)).toList();
+	}
+
+	private Object fromGenericMessage(GenericMessage<?> message, Class<?> targetClass,
+			@Nullable Object conversionHint) {
 		JsonNode jsonNode;
 		try {
 			jsonNode = this.jsonMapper.readTree(message.getPayload().toString());
@@ -77,7 +105,7 @@ public Object fromMessage(Message<?> message, Class<?> targetClass, @Nullable Ob
 				? ((SmartMessageConverter) this.payloadConverter).fromMessage(genericMessage, targetClass,
 						conversionHint)
 				: this.payloadConverter.fromMessage(genericMessage, targetClass);
-		return new SnsMessageWrapper(jsonNode.path("Subject").asText(), convertedMessage);
+		return convertedMessage;
 	}
 
 	@Override
@@ -97,10 +125,23 @@ public Message<?> toMessage(Object payload, MessageHeaders headers, Object conve
 				"This converter only supports reading a SNS notification and not writing them");
 	}
 
-	/**
-	 * SNS Message wrapper.
-	 */
-	public record SnsMessageWrapper(String subject, Object message) {
+	private static Type getResolvedType(Class<?> targetClass, @Nullable Object conversionHint) {
+		if (conversionHint instanceof MethodParameter param) {
+			param = param.nestedIfOptional();
+			if (Message.class.isAssignableFrom(param.getParameterType())) {
+				param = param.nested();
+			}
+			Type genericParameterType = param.getNestedGenericParameterType();
+			Class<?> contextClass = param.getContainingClass();
+			Type resolveType = GenericTypeResolver.resolveType(genericParameterType, contextClass);
+			if (resolveType instanceof ParameterizedType parameterizedType) {
+				return parameterizedType.getActualTypeArguments()[0];
+			}
+			else {
+				return resolveType;
+			}
+		}
+		return targetClass;
 	}
 
 }

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/support/resolver/NotificationMessageArgumentResolver.java

@@ -23,11 +23,11 @@
 import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.messaging.converter.SmartMessageConverter;
 import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver;
-import org.springframework.util.Assert;
 
 /**
  * @author Michael Sosa
  * @author gustavomonarin
+ * @author Wei Jiang
  * @since 3.1.1
  */
 public class NotificationMessageArgumentResolver implements HandlerMethodArgumentResolver {
@@ -45,10 +45,7 @@ public boolean supportsParameter(MethodParameter parameter) {
 
 	@Override
 	public Object resolveArgument(MethodParameter par, Message<?> msg) {
-		Object object = this.converter.fromMessage(msg, par.getParameterType(), par);
-		Assert.isInstanceOf(SnsMessageConverter.SnsMessageWrapper.class, object);
-		SnsMessageConverter.SnsMessageWrapper nr = (SnsMessageConverter.SnsMessageWrapper) object;
-		return nr.message();
+		return this.converter.fromMessage(msg, par.getParameterType(), par);
 	}
 
 }

spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsMessageConversionIntegrationTests.java

@@ -34,6 +34,8 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
@@ -56,6 +58,7 @@
  *
  * @author Tomaz Fernandes
  * @author Mikhail Strokov
+ * @author Wei Jiang
  */
 @SpringBootTest
 class SqsMessageConversionIntegrationTests extends BaseSqsIntegrationTest {
@@ -69,6 +72,7 @@ class SqsMessageConversionIntegrationTests extends BaseSqsIntegrationTest {
 	static final String RESOLVES_POJO_FROM_HEADER_QUEUE_NAME = "resolves_pojo_from_mapping_test_queue";
 	static final String RESOLVES_MY_OTHER_POJO_FROM_HEADER_QUEUE_NAME = "resolves_my_other_pojo_from_mapping_test_queue";
 	static final String RESOLVES_POJO_FROM_NOTIFICATION_MESSAGE_QUEUE_NAME = "resolves_pojo_from_notification_message_queue";
+	static final String RESOLVES_POJO_FROM_NOTIFICATION_MESSAGE_LIST_QUEUE_NAME = "resolves_pojo_from_notification_message_list_test_queue";
 
 	@Autowired
 	LatchContainer latchContainer;
@@ -85,7 +89,8 @@ static void beforeTests() {
 				createQueue(client, RESOLVES_POJO_MESSAGE_LIST_QUEUE_NAME),
 				createQueue(client, RESOLVES_POJO_FROM_HEADER_QUEUE_NAME),
 				createQueue(client, RESOLVES_MY_OTHER_POJO_FROM_HEADER_QUEUE_NAME),
-				createQueue(client, RESOLVES_POJO_FROM_NOTIFICATION_MESSAGE_QUEUE_NAME)).join();
+				createQueue(client, RESOLVES_POJO_FROM_NOTIFICATION_MESSAGE_QUEUE_NAME),
+				createQueue(client, RESOLVES_POJO_FROM_NOTIFICATION_MESSAGE_LIST_QUEUE_NAME)).join();
 	}
 
 	@Test
@@ -151,6 +156,18 @@ void resolvesMyPojoFromNotificationMessage() throws Exception {
 		assertThat(latchContainer.resolvesPojoNotificationMessageLatch.await(10, TimeUnit.SECONDS)).isTrue();
 	}
 
+	@Test
+	void resolvesMyPojoFromNotificationMessageList() throws Exception {
+		byte[] notificationJsonContent = FileCopyUtils
+				.copyToByteArray(getClass().getClassLoader().getResourceAsStream("notificationMessage.json"));
+		String payload = new String(notificationJsonContent);
+		List<Message<String>> messages = IntStream.range(0, 10).mapToObj(index -> MessageBuilder.withPayload(payload).build()).toList();
+		sqsTemplate.sendMany(RESOLVES_POJO_FROM_NOTIFICATION_MESSAGE_LIST_QUEUE_NAME, messages);
+		logger.debug("Sent message to queue {} with messageBody {}",
+				RESOLVES_POJO_FROM_NOTIFICATION_MESSAGE_LIST_QUEUE_NAME, payload);
+		assertThat(latchContainer.resolvesPojoNotificationMessageListLatch.await(10, TimeUnit.SECONDS)).isTrue();
+	}
+
 	private Map<String, Object> getHeaderMapping(Class<?> clazz) {
 		return Collections.singletonMap(SqsHeaders.SQS_DEFAULT_TYPE_HEADER, clazz.getName());
 	}
@@ -247,6 +264,14 @@ void listen(@SnsNotificationMessage MyEnvelope<MyPojo> myPojo) {
 					RESOLVES_POJO_FROM_NOTIFICATION_MESSAGE_QUEUE_NAME);
 			latchContainer.resolvesPojoNotificationMessageLatch.countDown();
 		}
+
+		@SqsListener(queueNames = RESOLVES_POJO_FROM_NOTIFICATION_MESSAGE_LIST_QUEUE_NAME, id = "resolves-pojo-with-notification-message-list", factory = "defaultSqsListenerContainerFactory")
+		void listen(@SnsNotificationMessage List<MyEnvelope<MyPojo>> myPojos) {
+			Assert.notEmpty(myPojos, "Received empty messages");
+			logger.debug("Received messages {} from queue {}", myPojos,
+					RESOLVES_POJO_FROM_NOTIFICATION_MESSAGE_LIST_QUEUE_NAME);
+			latchContainer.resolvesPojoNotificationMessageListLatch.countDown();
+		}
 	}
 
 	static class LatchContainer {
@@ -258,6 +283,7 @@ static class LatchContainer {
 		CountDownLatch resolvesPojoFromMappingLatch = new CountDownLatch(1);
 		CountDownLatch resolvesMyOtherPojoFromMappingLatch = new CountDownLatch(1);
 		CountDownLatch resolvesPojoNotificationMessageLatch = new CountDownLatch(1);
+		CountDownLatch resolvesPojoNotificationMessageListLatch = new CountDownLatch(1);
 
 	}