Sqs integration with sns notification message

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

@@ -17,17 +17,19 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
+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.MessageConverter;
+import org.springframework.messaging.converter.SmartMessageConverter;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.util.Assert;
 
 /**
  * @author Michael Sosa
  */
-public class SnsMessageConverter implements MessageConverter {
+public class SnsMessageConverter implements SmartMessageConverter {
 
  private final ObjectMapper jsonMapper;
 
@@ -41,7 +43,7 @@ public SnsMessageConverter(MessageConverter payloadConverter, ObjectMapper jsonM
  }
 
  @Override
- public Object fromMessage(Message<?> message, Class<?> targetClass) {
+ 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");
 
@@ -69,8 +71,15 @@ public Object fromMessage(Message<?> message, Class<?> targetClass) {
 
  String messagePayload = jsonNode.get("Message").asText();
  GenericMessage<String> genericMessage = new GenericMessage<>(messagePayload);
- return new SnsMessageWrapper(jsonNode.path("Subject").asText(),
- this.payloadConverter.fromMessage(genericMessage, targetClass));
+ Object convertedMessage = (payloadConverter instanceof SmartMessageConverter) ?
+ ((SmartMessageConverter)this.payloadConverter).fromMessage(genericMessage, targetClass, conversionHint)
+ : this.payloadConverter.fromMessage(genericMessage, targetClass);
+ return new SnsMessageWrapper(jsonNode.path("Subject").asText(), convertedMessage);
+ }
+
+ @Override
+ public Object fromMessage(Message<?> message, Class<?> targetClass) {
+ return fromMessage( message, targetClass, null);
  }
 
  @Override
@@ -79,6 +88,12 @@ public Message<?> toMessage(Object payload, MessageHeaders headers) {
  "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");
+ }
+
  /**
  * SNS Message wrapper.
  */

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

@@ -21,6 +21,7 @@
 import org.springframework.core.MethodParameter;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.converter.MessageConverter;
+import org.springframework.messaging.converter.SmartMessageConverter;
 import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver;
 import org.springframework.util.Assert;
 
@@ -29,7 +30,7 @@
  */
 public class NotificationMessageArgumentResolver implements HandlerMethodArgumentResolver {
 
- private final MessageConverter converter;
+ private final SmartMessageConverter converter;
 
  public NotificationMessageArgumentResolver(MessageConverter converter, ObjectMapper jsonMapper) {
  this.converter = new SnsMessageConverter(converter, jsonMapper);
@@ -41,8 +42,8 @@ public boolean supportsParameter(MethodParameter parameter) {
  }
 
  @Override
- public Object resolveArgument(MethodParameter par, Message<?> msg) throws Exception {
- Object object = this.converter.fromMessage(msg, par.getParameterType());
+ 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();

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

@@ -241,9 +241,10 @@ static class ResolvesPojoWithNotificationAnnotationListener {
  LatchContainer latchContainer;
 
  @SqsListener(queueNames = RESOLVES_POJO_FROM_NOTIFICATION_MESSAGE_QUEUE_NAME, id = "resolves-pojo-with-notification-message", factory = "defaultSqsListenerContainerFactory")
- void listen(@SnsNotificationMessage MyPojo myPojo) {
- Assert.notNull((myPojo).firstField, "Received null message");
- logger.debug("Received message {} from queue {}", myPojo, RESOLVES_POJO_FROM_NOTIFICATION_MESSAGE_QUEUE_NAME);
+ void listen(@SnsNotificationMessage MyEnvelope<MyPojo> myPojo) {
+ Assert.notNull((myPojo).data.firstField, "Received null message");
+ logger.debug("Received message {} from queue {}", myPojo,
+ RESOLVES_POJO_FROM_NOTIFICATION_MESSAGE_QUEUE_NAME);
  latchContainer.resolvesPojoNotificationMessageLatch.countDown();
  }
  }
@@ -341,6 +342,7 @@ ResolvesMyPojoWithMappingListener resolvesMyPojoWithMappingListener() {
  ResolvesMyOtherPojoWithMappingListener resolvesMyOtherPojoWithMappingListener() {
  return new ResolvesMyOtherPojoWithMappingListener();
  }
+
  @Bean
  ResolvesPojoWithNotificationAnnotationListener resolvesPojoWithNotificationAnnotationListener() {
  return new ResolvesPojoWithNotificationAnnotationListener();
@@ -363,6 +365,27 @@ SqsTemplate sqsTemplate() {
 
  }
 
+ static class MyEnvelope<T> {
+ String specversion;
+ T data;
+
+ public String getSpecversion() {
+ return specversion;
+ }
+
+ public void setSpecversion(String specversion) {
+ this.specversion = specversion;
+ }
+
+ public T getData() {
+ return data;
+ }
+
+ public void setData(T data) {
+ this.data = data;
+ }
+ }
+
  static class MyPojo implements MyInterface {
 
  String firstField;

spring-cloud-aws-sqs/src/test/resources/notificationMessage.json

@@ -2,7 +2,7 @@
  "Type": "Notification",
  "MessageId": "f2c15fec-c617-5b08-b54d-13c4099fec60",
  "TopicArn": "arn:aws:sns:eu-west-1:111111111111:mySampleTopic",
- "Message": "{\"firstField\":\"pojoNotificationMessage\",\"secondField\":\"secondValue\"}",
+ "Message": "{\"specversion\": \"1.0.2\", \"data\": {\"firstField\":\"pojoNotificationMessage\",\"secondField\":\"secondValue\"}}",
  "Timestamp": "2014-06-28T14:12:24.418Z",
  "SignatureVersion": "1",
  "Signature": "XDvKSAnhxECrAmyIrs0Dsfbp/tnKD1IvoOOYTU28FtbUoxr/CgziuW87yZwTuSNNbHJbdD3BEjHS0vKewm0xBeQ0PToDkgtoORXo5RWnmShDQ2nhkthFhZnNulKtmFtRogjBtCwbz8sPnbOCSk21ruyXNdV2RUbdDalndAW002CWEQmYMxFSN6OXUtMueuT610aX+tqeYP4Z6+8WTWLWjAuVyy7rOI6KHYBcVDhKtskvTOPZ4tiVohtQdQbO2Gjuh1vblRzzwMkfaoFTSWImd4pFXxEsv/fq9aGIlqq9xEryJ0w2huFwI5gxyhvGt0RnTd9YvmAEC+WzdJDOqaDNxg==",