Add SQSListener argument resolver to extract the subject of an SNS message

Author: n-n-nebbl

spring-cloud-aws-autoconfigure/src/main/java/io/awspring/cloud/autoconfigure/config/ConfigStoreRuntimeHints.java

@@ -52,12 +52,12 @@ public void registerHints(RuntimeHints hints, ClassLoader classLoader) {
 
 		if (ClassUtils.isPresent("io.awspring.cloud.s3.S3PropertySource", classLoader)) {
 			hints.reflection().registerType(TypeReference.of(S3PropertySources.class),
-				hint -> hint.withMembers(MemberCategory.INVOKE_DECLARED_CONSTRUCTORS,
-					MemberCategory.INTROSPECT_DECLARED_METHODS, MemberCategory.DECLARED_FIELDS));
+					hint -> hint.withMembers(MemberCategory.INVOKE_DECLARED_CONSTRUCTORS,
+							MemberCategory.INTROSPECT_DECLARED_METHODS, MemberCategory.DECLARED_FIELDS));
 
 			hints.reflection().registerType(TypeReference.of(S3PropertySource.class),
-				hint -> hint.withMembers(MemberCategory.INVOKE_DECLARED_CONSTRUCTORS,
-					MemberCategory.INTROSPECT_DECLARED_METHODS, MemberCategory.DECLARED_FIELDS));
+					hint -> hint.withMembers(MemberCategory.INVOKE_DECLARED_CONSTRUCTORS,
+							MemberCategory.INTROSPECT_DECLARED_METHODS, MemberCategory.DECLARED_FIELDS));
 		}
 	}
 }

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

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2013-2022 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.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Annotation that is used to map SNS notification subject on an SQS Queue to a variable that is annotated. Used in
+ * Controllers method for handling/receiving SQS notifications.
+ *
+ * @author Alexander Nebel
+ * @since 3.3.0
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.PARAMETER)
+public @interface SnsNotificationSubject {
+
+}

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

@@ -22,6 +22,7 @@
 import io.awspring.cloud.sqs.listener.SqsHeaders;
 import io.awspring.cloud.sqs.support.resolver.BatchVisibilityHandlerMethodArgumentResolver;
 import io.awspring.cloud.sqs.support.resolver.NotificationMessageArgumentResolver;
+import io.awspring.cloud.sqs.support.resolver.NotificationSubjectArgumentResolver;
 import io.awspring.cloud.sqs.support.resolver.QueueAttributesMethodArgumentResolver;
 import io.awspring.cloud.sqs.support.resolver.SqsMessageMethodArgumentResolver;
 import io.awspring.cloud.sqs.support.resolver.VisibilityHandlerMethodArgumentResolver;
@@ -84,6 +85,7 @@ protected Collection<HandlerMethodArgumentResolver> createAdditionalArgumentReso
 		List<HandlerMethodArgumentResolver> argumentResolvers = new ArrayList<>(createAdditionalArgumentResolvers());
 		if (objectMapper != null) {
 			argumentResolvers.add(new NotificationMessageArgumentResolver(messageConverter, objectMapper));
+			argumentResolvers.add(new NotificationSubjectArgumentResolver(objectMapper));
 		}
 		return argumentResolvers;
 	}

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

@@ -0,0 +1,97 @@
+/*
+ * Copyright 2013-2023 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.support.converter;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.List;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.converter.MessageConversionException;
+import org.springframework.messaging.converter.SmartMessageConverter;
+import org.springframework.util.Assert;
+
+/**
+ * @author Alexander Nebel
+ * @since 3.3.0
+ */
+public class SnsSubjectConverter implements SmartMessageConverter {
+
+	private final ObjectMapper jsonMapper;
+
+	public SnsSubjectConverter(ObjectMapper jsonMapper) {
+		Assert.notNull(jsonMapper, "jsonMapper must not be null");
+		this.jsonMapper = jsonMapper;
+	}
+
+	@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 (!targetClass.isAssignableFrom(String.class)) {
+			throw new MessageConversionException("Subject can only be injected into String assignable Types", null);
+		}
+		if (payload instanceof List) {
+			throw new MessageConversionException("Conversion of List is not supported", null);
+		}
+
+		JsonNode jsonNode;
+		try {
+			jsonNode = this.jsonMapper.readTree(message.getPayload().toString());
+		}
+		catch (Exception e) {
+			throw new MessageConversionException("Could not read JSON", e);
+		}
+		if (!jsonNode.has("Type")) {
+			throw new MessageConversionException(
+					"Payload: '" + message.getPayload() + "' does not contain a Type attribute", null);
+		}
+
+		if (!"Notification".equals(jsonNode.get("Type").asText())) {
+			throw new MessageConversionException("Payload: '" + message.getPayload() + "' is not a valid notification",
+					null);
+		}
+
+		if (!jsonNode.has("Subject")) {
+			throw new MessageConversionException("Payload: '" + message.getPayload() + "' does not contain a subject",
+					null);
+		}
+
+		return jsonNode.get("Subject").asText();
+	}
+
+	@Override
+	public Object fromMessage(Message<?> message, Class<?> targetClass) {
+		return fromMessage(message, targetClass, null);
+	}
+
+	@Override
+	public Message<?> toMessage(Object payload, MessageHeaders headers) {
+		throw new UnsupportedOperationException(
+				"This converter only supports reading a SNS notification and not writing them");
+	}
+
+	@Override
+	public Message<?> toMessage(Object payload, MessageHeaders headers, Object conversionHint) {
+		throw new UnsupportedOperationException(
+				"This converter only supports reading a SNS notification and not writing them");
+	}
+}

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

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2013-2023 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.support.resolver;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.awspring.cloud.sqs.annotation.SnsNotificationSubject;
+import io.awspring.cloud.sqs.support.converter.SnsSubjectConverter;
+import org.springframework.core.MethodParameter;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.converter.SmartMessageConverter;
+import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver;
+
+/**
+ * @author Alexander Nebel
+ * @since 3.3.0
+ */
+public class NotificationSubjectArgumentResolver implements HandlerMethodArgumentResolver {
+
+	private final SmartMessageConverter converter;
+
+	public NotificationSubjectArgumentResolver(ObjectMapper jsonMapper) {
+		this.converter = new SnsSubjectConverter(jsonMapper);
+	}
+
+	@Override
+	public boolean supportsParameter(MethodParameter parameter) {
+		return parameter.hasParameterAnnotation(SnsNotificationSubject.class);
+	}
+
+	@Override
+	public Object resolveArgument(MethodParameter par, Message<?> msg) {
+		return converter.fromMessage(msg, par.getParameterType(), par);
+	}
+
+}