Merge pull request #101 from JaidenAshmore/issue/92_message_attribute_argument_resolver

refs #92: Message Attribute support

Author: JaidenAshmore

doc/core-implementations-overview.md

@@ -72,6 +72,11 @@ with an implementation that allows for a message to be manually acknowledged whe
 signature, the [MessageProcessor](../java-dynamic-sqs-listener-api/src/main/java/com/jashmore/sqs/processor/MessageProcessor.java) is not required to
 acknowledge the message after a successful execution. These implementations should be provided by the
 [MessageProcessor](../java-dynamic-sqs-listener-api/src/main/java/com/jashmore/sqs/processor/MessageProcessor.java) being used.
+- [MessageAttribute](../java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/argument/attribute/MessageAttribute.java): arguments annotated with this
+will attempt to parse the contents of the message attribute into this field. For example, if the argument is a String then the attribute will be cast to a
+string where as if the argument is an integer it will try and parse the string into the number.  This also works with POJOs in that the resolver will
+ attempt to deserialised the message attribute into this POJO shape, e.g. via the Jackson Object Mapper.  This is provided by the
+[MessageAttributeArgumentResolver](../java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/argument/attribute/MessageAttributeArgumentResolver.java).
 - [VisibilityExtender](../java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/argument/visibility/VisibilityExtender.java): arguments of this type
 will be injected with an implementation that extends the message visibility of the current message. This is provided by the
 [VisibilityExtenderArgumentResolver](../java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/argument/visibility/VisibilityExtenderArgumentResolver.java). 

doc/documentation.md

@@ -19,6 +19,8 @@ more in depth understanding take a look at the JavaDoc for the API.
         1. [How to add a custom ArgumentResolver to a Spring application](how-to-guides/spring/spring-how-to-add-custom-argument-resolver.md): useful for
         integrating custom argument resolution code to be included in a Spring Application. See [How to implement a custom ArgumentResolver](how-to-guides/core/core-how-to-implement-a-custom-argument-resolver.md)
         for how build a new ArgumentResolver from scratch
+        1. [How to provide a custom Object Mapper](how-to-guides/spring/spring-how-to-add-custom-argument-resolver.md): guide for overriding the default
+        `ObjectMapper` that is used to serialise the message body and attributes
         1. [How to add your own queue listener](how-to-guides/spring/spring-how-to-add-own-queue-listener.md): useful for defining your own annotation for
         queue listening without the verbosity of the custom queue listener
         1. [How to write Spring Integration Tests](how-to-guides/spring/spring-how-to-write-integration-tests.md): you actually want to test what you are

doc/how-to-guides/spring/spring-how-to-provide-a-custom-object-mapper.md

@@ -0,0 +1,45 @@
+# Spring - How to provide own ObjectMapper for serialisation/deserialisation
+Currently a Jackson `ObjectMapper` is being used for the serialisation and deserialisation of messages and their attributes. If none is provided
+by the application, the Spring Starter will provide their own to use. This shows steps on providing your own so the default is not used.
+
+### Steps
+
+1. In a `@Configuration` class define a bean of type `ObjectMapper`
+    ```java
+    @Configuration
+    public class MyConfiguration {
+       @Bean
+       public ObjectMapper objectMapper() {
+           return new ObjectMapper();
+       }     
+    }
+    ```
+
+
+## Providing an ObjectMapper without including it in the Spring Context
+One problem with doing this method is that this ObjectMapper will be shared with this library as well as any others that need this, e.g. the HTTP layer
+may be using this ObjectMapper for their own serialisation. Therefore, if you want to provide an `ObjectMapper` that is only used by this library you would
+need to override the individual core [ArgumentResolvers](../../../java-dynamic-sqs-listener-api/src/main/java/com/jashmore/sqs/argument/ArgumentResolver.java)
+that are using this `ObjectMapper`, which is the
+[PayloadArgumentResolver](../../../java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/argument/payload/PayloadArgumentResolver.java) and
+[MessageAttributeArgumentResolver](../../../java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/argument/attribute/MessageAttributeArgumentResolver.java). 
+
+### Steps
+
+1. In a `@Configuration` class define your own [ArgumentResolverService](../../../java-dynamic-sqs-listener-api/src/main/java/com/jashmore/sqs/argument/ArgumentResolverService.java)
+    ```java
+    @Configuration
+    public class MyConfiguration {
+       private static final ObjectMapper MY_OBJECT_MAPPER_FOR_SQS = new ObjectMapper();
+    
+       @Bean
+       public PayloadArgumentResolver payloadArgumentResolver() {
+           return new PayloadArgumentResolver(new JacksonPayloadMapper(MY_OBJECT_MAPPER_FOR_SQS));
+       }   
+    
+       @Bean
+       public MessageAttributeArgumentResolver messageAttributeArgumentResolver() {
+           return new MessageAttributeArgumentResolver(objectMapper);
+       }
+    }
+    ```

examples/java-dynamic-sqs-listener-core-examples/src/main/java/com/jashmore/sqs/examples/ConcurrentBrokerExample.java

@@ -189,7 +189,7 @@ private static SqsAsyncClient startElasticMqServer() throws URISyntaxException {
      */
     private static ArgumentResolverService argumentResolverService(final SqsAsyncClient sqsAsyncClient) {
         final PayloadMapper payloadMapper = new JacksonPayloadMapper(OBJECT_MAPPER);
-        return new CoreArgumentResolverService(payloadMapper, sqsAsyncClient);
+        return new CoreArgumentResolverService(payloadMapper, sqsAsyncClient, OBJECT_MAPPER);
     }
 
     /**

java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/argument/CoreArgumentResolverService.java

@@ -2,6 +2,8 @@
 
 import com.google.common.collect.ImmutableSet;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.jashmore.sqs.argument.attribute.MessageAttributeArgumentResolver;
 import com.jashmore.sqs.argument.messageid.MessageIdArgumentResolver;
 import com.jashmore.sqs.argument.payload.PayloadArgumentResolver;
 import com.jashmore.sqs.argument.payload.mapper.PayloadMapper;
@@ -22,10 +24,12 @@ public class CoreArgumentResolverService implements ArgumentResolverService {
     private final DelegatingArgumentResolverService delegatingArgumentResolverService;
 
     public CoreArgumentResolverService(final PayloadMapper payloadMapper,
-                                       final SqsAsyncClient sqsAsyncClient) {
+                                       final SqsAsyncClient sqsAsyncClient,
+                                       final ObjectMapper objectMapper) {
         final Set<ArgumentResolver<?>> argumentResolvers = ImmutableSet.of(
                 new PayloadArgumentResolver(payloadMapper),
                 new MessageIdArgumentResolver(),
+                new MessageAttributeArgumentResolver(objectMapper),
                 new VisibilityExtenderArgumentResolver(sqsAsyncClient)
         );
         this.delegatingArgumentResolverService = new DelegatingArgumentResolverService(argumentResolvers);

java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/argument/DelegatingArgumentResolverService.java

@@ -18,21 +18,21 @@ public class DelegatingArgumentResolverService implements ArgumentResolverServic
 
     @Override
     public Object resolveArgument(final QueueProperties queueProperties, final MethodParameter methodParameter, final Message message) {
-        return argumentResolvers.stream()
-                .filter(resolver -> resolver.canResolveParameter(methodParameter))
-                .map(resolver -> {
-                    try {
-                        return resolver.resolveArgumentForParameter(queueProperties, methodParameter, message);
-                    } catch (final RuntimeException runtimeException) {
-                        // Make sure to wrap any unintended exceptions with the expected exception for errors
-                        if (!ArgumentResolutionException.class.isAssignableFrom(runtimeException.getClass())) {
-                            throw new ArgumentResolutionException("Error obtaining an argument value for parameter", runtimeException);
-                        }
-
-                        throw runtimeException;
+        for (final ArgumentResolver<?> resolver: argumentResolvers) {
+            if (resolver.canResolveParameter(methodParameter)) {
+                try {
+                    return resolver.resolveArgumentForParameter(queueProperties, methodParameter, message);
+                } catch (final RuntimeException runtimeException) {
+                    // Make sure to wrap any unintended exceptions with the expected exception for errors
+                    if (!ArgumentResolutionException.class.isAssignableFrom(runtimeException.getClass())) {
+                        throw new ArgumentResolutionException("Error obtaining an argument value for parameter", runtimeException);
                     }
-                })
-                .findFirst()
-                .orElseThrow(() -> new ArgumentResolutionException("No ArgumentResolver found that can process this parameter"));
+
+                    throw runtimeException;
+                }
+            }
+        }
+
+        throw new ArgumentResolutionException("No ArgumentResolver found that can process this parameter");
     }
 }

java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/argument/attribute/MessageAttribute.java

@@ -0,0 +1,28 @@
+package com.jashmore.sqs.argument.attribute;
+
+import static java.lang.annotation.ElementType.PARAMETER;
+import static java.lang.annotation.RetentionPolicy.RUNTIME;
+
+import java.lang.annotation.Retention;
+import java.lang.annotation.Target;
+
+/**
+ * Populate on of the arguments in a message processing method by taking a value from the message attributes.
+ */
+@Retention(RUNTIME)
+@Target(PARAMETER)
+public @interface MessageAttribute {
+    /**
+     * The name of the attribute to use.
+     *
+     * @return the attribute name
+     */
+    String value();
+
+    /**
+     * Fail the processing of the message if the message attribute with the provided name does not exist.
+     *
+     * @return whether the message should fail to be processed if the message attribute is missing
+     */
+    boolean required() default false;
+}

java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/argument/attribute/MessageAttributeArgumentResolver.java

@@ -0,0 +1,121 @@
+package com.jashmore.sqs.argument.attribute;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.jashmore.sqs.QueueProperties;
+import com.jashmore.sqs.argument.ArgumentResolutionException;
+import com.jashmore.sqs.argument.ArgumentResolver;
+import com.jashmore.sqs.argument.MethodParameter;
+import com.jashmore.sqs.util.annotation.AnnotationUtils;
+import software.amazon.awssdk.services.sqs.model.Message;
+import software.amazon.awssdk.services.sqs.model.MessageAttributeValue;
+
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.util.Optional;
+
+/**
+ * An {@link ArgumentResolver} that is able to handle the extraction of information from the attributes of the SQS message.
+ *
+ * <p>This will attempt to do its best in casting the contents of the message attribute to the type of tha parameter. For example, if the contents of
+ * the message attribute is binary (byte[]) and the parameter is a POJO, it will attempt to serialise using the {@link ObjectMapper#readValue(byte[], Class)}.
+ *
+ * <p>It is the responsibility of the consumer to make sure that the type of the parameter is correct in regards to the content of the message attribute. For
+ * example, this resolver ignores all helper data types after the main, e.g. Number.float data types will have the float ignored.
+ *
+ * <p>This current implementation uses the Jackson {@link ObjectMapper} to perform all of the parsing and there is the potential for a future version of this
+ * library to split this out so that Jackson isn't a required dependency.
+ *
+ * @see <a href="https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-message-attributes.html">SQS Message Attributes</a>
+ * @see MessageAttributeValue
+ */
+public class MessageAttributeArgumentResolver implements ArgumentResolver<Object> {
+    private final ObjectMapper objectMapper;
+
+    public MessageAttributeArgumentResolver(final ObjectMapper objectMapper) {
+        this.objectMapper = objectMapper;
+    }
+
+    @Override
+    public boolean canResolveParameter(final MethodParameter methodParameter) {
+        return AnnotationUtils.findParameterAnnotation(methodParameter, MessageAttribute.class).isPresent();
+    }
+
+    @Override
+    public Object resolveArgumentForParameter(final QueueProperties queueProperties,
+                                              final MethodParameter methodParameter,
+                                              final Message message) throws ArgumentResolutionException {
+        final MessageAttribute annotation = AnnotationUtils.findParameterAnnotation(methodParameter, MessageAttribute.class)
+                .orElseThrow(() -> new ArgumentResolutionException("Parameter passed in does not contain the MessageAttribute annotation when it should"));
+
+        final String attributeName = annotation.value();
+
+        final Optional<MessageAttributeValue> optionalMessageAttributeValue = Optional.ofNullable(message.messageAttributes().get(attributeName));
+
+        if (!optionalMessageAttributeValue.isPresent()) {
+            if (annotation.required()) {
+                throw new ArgumentResolutionException("Required Message Attribute '" + attributeName + "' is missing from message");
+            }
+
+            return null;
+        }
+
+        final MessageAttributeValue messageAttributeValue = optionalMessageAttributeValue.get();
+
+        if (messageAttributeValue.dataType().startsWith(MessageAttributeDataTypes.STRING.getValue())
+                || messageAttributeValue.dataType().startsWith(MessageAttributeDataTypes.NUMBER.getValue())) {
+            return handleStringParameterValue(methodParameter, messageAttributeValue, attributeName);
+        } else if (messageAttributeValue.dataType().startsWith(MessageAttributeDataTypes.BINARY.getValue())) {
+            return handleByteParameterValue(methodParameter, messageAttributeValue);
+        }
+
+        throw new ArgumentResolutionException("Cannot parse message attribute due to unknown data type '" + messageAttributeValue.dataType() + "'");
+    }
+
+    /**
+     * Handle resolving the argument from the string contents of the attribute.
+     *
+     * @param methodParameter       the parameter of the method to resolve
+     * @param messageAttributeValue the value of the message attribute
+     * @param attributeName         the name of the attribute that is being consumed
+     * @return the resolved argument from the attribute
+     */
+    private Object handleStringParameterValue(final MethodParameter methodParameter,
+                                              final MessageAttributeValue messageAttributeValue,
+                                              final String attributeName) {
+        if (methodParameter.getParameter().getType().isAssignableFrom(String.class)) {
+            return messageAttributeValue.stringValue();
+        }
+
+        try {
+            return objectMapper.readValue(messageAttributeValue.stringValue(), methodParameter.getParameter().getType());
+        } catch (final IOException ioException) {
+            throw new ArgumentResolutionException("Error parsing Message Attribute '" + attributeName + "'", ioException);
+        }
+    }
+
+    /**
+     * Handle an attribute that contains the data as bytes.
+     *
+     * @param methodParameter       details about the parameter to parse the message attribute into
+     * @param messageAttributeValue value of the message attribute
+     * @return the argument resolved for this attribute
+     */
+    private Object handleByteParameterValue(final MethodParameter methodParameter,
+                                            final MessageAttributeValue messageAttributeValue) {
+        final byte[] byteArray = messageAttributeValue.binaryValue().asByteArray();
+        final Class<?> parameterClass = methodParameter.getParameter().getType();
+        if (parameterClass == byte[].class) {
+            return byteArray;
+        }
+
+        if (parameterClass.isAssignableFrom(String.class)) {
+            return new String(byteArray, Charset.forName("UTF-8"));
+        }
+
+        try {
+            return objectMapper.readValue(byteArray, parameterClass);
+        } catch (final IOException ioException) {
+            throw new ArgumentResolutionException("Failure to parse binary bytes to '" + parameterClass.getName() + "'", ioException);
+        }
+    }
+}

java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/argument/attribute/MessageAttributeDataTypes.java

@@ -0,0 +1,14 @@
+package com.jashmore.sqs.argument.attribute;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+@Getter
+@AllArgsConstructor
+public enum MessageAttributeDataTypes {
+    STRING("String"),
+    NUMBER("Number"),
+    BINARY("Binary");
+
+    private final String value;
+}

java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/broker/concurrent/ConcurrentMessageBroker.java

@@ -68,12 +68,17 @@ public void run() {
                     try {
                         return messageRetriever.retrieveMessage();
                     } catch (final InterruptedException exception) {
-                        log.debug("Thread interrupted waiting for a message");
-                        throw new RuntimeException("Failure to get message");
+                        log.trace("Thread interrupted waiting for a message");
+                        throw new BrokerStoppedWhileRetrievingMessageException();
                     }
                 }, concurrentThreadsExecutor)
                         .thenAcceptAsync(messageProcessor::processMessage, messageProcessingThreadsExecutor)
-                        .whenComplete((ignoredResult, throwable) -> concurrentMessagesBeingProcessedSemaphore.release());
+                        .whenComplete((ignoredResult, throwable) -> {
+                            if (throwable != null && !(throwable.getCause() instanceof BrokerStoppedWhileRetrievingMessageException)) {
+                                log.error("Error processing message", throwable.getCause());
+                            }
+                            concurrentMessagesBeingProcessedSemaphore.release();
+                        });
             } catch (final Throwable throwable) {
                 try {
                     final long errorBackoffTimeInMilliseconds = getErrorBackoffTimeInMilliseconds();
@@ -176,4 +181,12 @@ private long getNumberOfMillisecondsToObtainPermit() {
                 DEFAULT_BACKOFF_TIME_IN_MS
         );
     }
+
+    /**
+     * Internal exception used to be thrown when the thread is interrupted while retrieving messages. This is because we don't want a
+     * error to be logged for this scenario but only when their was an actual exception processing the message.
+     */
+    private static class BrokerStoppedWhileRetrievingMessageException extends RuntimeException {
+
+    }
 }