From d61525e54b84fcebfc3783b81baf2769d56371d9 Mon Sep 17 00:00:00 2001 From: msosa Date: Tue, 26 Sep 2023 16:46:46 -0400 Subject: [PATCH] Add `NotificationRequestConverter` and `NotificationMessageArgumentResolver` to allow reading SNS messages on SQS --- ...ctListenerAnnotationBeanPostProcessor.java | 2 + .../annotation/SnsNotificationMessage.java | 33 ++++++ .../NotificationRequestConverter.java | 103 ++++++++++++++++++ .../NotificationMessageArgumentResolver.java | 49 +++++++++ 4 files changed, 187 insertions(+) create mode 100644 spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/annotation/SnsNotificationMessage.java create mode 100644 spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/support/converter/NotificationRequestConverter.java create mode 100644 spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/support/resolver/NotificationMessageArgumentResolver.java diff --git a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/annotation/AbstractListenerAnnotationBeanPostProcessor.java b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/annotation/AbstractListenerAnnotationBeanPostProcessor.java index 55f15af67..a24c5209f 100644 --- a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/annotation/AbstractListenerAnnotationBeanPostProcessor.java +++ b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/annotation/AbstractListenerAnnotationBeanPostProcessor.java @@ -25,6 +25,7 @@ import io.awspring.cloud.sqs.support.resolver.AcknowledgmentHandlerMethodArgumentResolver; import io.awspring.cloud.sqs.support.resolver.BatchAcknowledgmentArgumentResolver; import io.awspring.cloud.sqs.support.resolver.BatchPayloadMethodArgumentResolver; +import io.awspring.cloud.sqs.support.resolver.NotificationMessageArgumentResolver; import java.lang.annotation.Annotation; import java.lang.reflect.Method; import java.util.ArrayList; @@ -290,6 +291,7 @@ protected List createArgumentResolvers(MessageCon new HeadersMethodArgumentResolver(), new BatchPayloadMethodArgumentResolver(messageConverter, this.endpointRegistrar.getValidator()), new MessageMethodArgumentResolver(messageConverter), + new NotificationMessageArgumentResolver(messageConverter), new PayloadMethodArgumentResolver(messageConverter, this.endpointRegistrar.getValidator())); } // @formatter:on diff --git a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/annotation/SnsNotificationMessage.java b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/annotation/SnsNotificationMessage.java new file mode 100644 index 000000000..058db55a9 --- /dev/null +++ b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/annotation/SnsNotificationMessage.java @@ -0,0 +1,33 @@ +/* + * 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 value on an SQS Queue to a variable that is annotated. + * Used in Controllers method for handling/receiving SQS notifications. + * + * @author Michael Sosa + */ +@Retention(RetentionPolicy.RUNTIME) +@Target(ElementType.PARAMETER) +public @interface SnsNotificationMessage { + +} diff --git a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/support/converter/NotificationRequestConverter.java b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/support/converter/NotificationRequestConverter.java new file mode 100644 index 000000000..e13ef1c15 --- /dev/null +++ b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/support/converter/NotificationRequestConverter.java @@ -0,0 +1,103 @@ +/* + * 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 org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.converter.MessageConversionException; +import org.springframework.messaging.converter.MessageConverter; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.util.Assert; + +/** + * @author Michael Sosa + */ +public class NotificationRequestConverter implements MessageConverter { + + private final ObjectMapper jsonMapper = new ObjectMapper(); + + private final MessageConverter payloadConverter; + + public NotificationRequestConverter(MessageConverter payloadConverter) { + this.payloadConverter = payloadConverter; + } + @Override + public Object fromMessage(Message message, Class targetClass) { + Assert.notNull(message, "message must not be null"); + Assert.notNull(targetClass, "target class must not be 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("Message")) { + throw new MessageConversionException("Payload: '" + message.getPayload() + "' does not contain a message", + null); + } + + String messagePayload = jsonNode.get("Message").asText(); + GenericMessage genericMessage = new GenericMessage<>(messagePayload); + return new NotificationRequest(jsonNode.path("Subject").asText(), + this.payloadConverter.fromMessage(genericMessage, targetClass)); + } + + @Override + public Message toMessage(Object payload, MessageHeaders headers) { + throw new UnsupportedOperationException( + "This converter only supports reading a SNS notification and not writing them"); + } + + /** + * Notification request wrapper. + */ + public static class NotificationRequest { + + private final String subject; + + private final Object message; + + public NotificationRequest(String subject, Object message) { + this.subject = subject; + this.message = message; + } + + public String getSubject() { + return this.subject; + } + + public Object getMessage() { + return this.message; + } + + } + +} diff --git a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/support/resolver/NotificationMessageArgumentResolver.java b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/support/resolver/NotificationMessageArgumentResolver.java new file mode 100644 index 000000000..534026e83 --- /dev/null +++ b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/support/resolver/NotificationMessageArgumentResolver.java @@ -0,0 +1,49 @@ +/* + * 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 io.awspring.cloud.sqs.annotation.SnsNotificationMessage; +import io.awspring.cloud.sqs.support.converter.NotificationRequestConverter; +import org.springframework.core.MethodParameter; +import org.springframework.messaging.Message; +import org.springframework.messaging.converter.MessageConverter; +import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver; + +/** + * @author Michael Sosa + */ +public class NotificationMessageArgumentResolver implements HandlerMethodArgumentResolver { + + private final MessageConverter converter; + + public NotificationMessageArgumentResolver(MessageConverter converter) { + this.converter = new NotificationRequestConverter(converter); + } + + @Override + public boolean supportsParameter(MethodParameter parameter) { + return parameter.hasParameterAnnotation(SnsNotificationMessage.class); + } + + @Override + public Object resolveArgument(MethodParameter par, Message msg) throws Exception { + Object object = this.converter.fromMessage(msg, par.getParameterType()); + NotificationRequestConverter.NotificationRequest nr = (NotificationRequestConverter.NotificationRequest) object; + return nr.getMessage(); + } + +}