Merge pull request #95 from JaidenAshmore/issue/90_spring_properties_in_annotations

refs 90: allow for Spring properties to be included in Queue Annotations

Author: JaidenAshmore

doc/how-to-guides/spring/spring-how-to-set-queue-listener-props-from-the-environment.md

@@ -0,0 +1,31 @@
+# Spring - How to set Queue Listener props from the Environment
+The Queue Listener annotations have been written to allow for them to be set via the Spring Environment, e.g. properties from an `application.yml` file.
+
+# Steps
+
+1. Create some properties in the `application.yml` file, in this case there are two profiles with different concurrency rates.
+    ```yaml
+    
+    ---
+    spring.profiles: staging
+    
+    queues:
+        my-queue:
+           concurrency: 5
+ 
+    ---
+    spring.profiles: production
+    queues:
+        my-queue:
+           concurrency: 5
+    ```
+1. Use the `{propertyName}String` fields in the annotations to pull data from the environment:
+    ```java
+    @QueueListener(value = "http://localhost:9432/q/myqueue", concurrencyLevelString="${queues.my-queue.concurrency}")
+    public void myMethod(@Payload String payload) {
+    
+    }
+    ```
+    
+***NOTE**: the `{propertyName}String` fields will override any of the other properties if they are set. E.g. if both `concurrencyLevel` and
+`concurrencyLevelString` are set the `concurrencyLevelString` one will be prioritised*

java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/retriever/batching/StaticBatchingMessageRetrieverProperties.java

@@ -1,6 +1,7 @@
 package com.jashmore.sqs.retriever.batching;
 
 import lombok.Builder;
+import lombok.Value;
 
 import javax.annotation.Nullable;
 import javax.validation.constraints.Positive;
@@ -9,6 +10,7 @@
 /**
  * Static implementation of the properties that will never change during the processing of the messages.
  */
+@Value
 @Builder(toBuilder = true)
 public class StaticBatchingMessageRetrieverProperties implements BatchingMessageRetrieverProperties {
     private final int numberOfThreadsWaitingTrigger;

java-dynamic-sqs-listener-spring/java-dynamic-sqs-listener-spring-starter/src/main/java/com/jashmore/sqs/spring/config/QueueListenerConfiguration.java

@@ -173,22 +173,25 @@ public static class QueueWrapperConfiguration {
             @Bean
             public QueueWrapper coreProvidedQueueListenerWrapper(final ArgumentResolverService argumentResolverService,
                                                                  final SqsAsyncClient sqsAsyncClient,
-                                                                 final QueueResolverService queueResolverService) {
-                return new QueueListenerWrapper(argumentResolverService, sqsAsyncClient, queueResolverService);
+                                                                 final QueueResolverService queueResolverService,
+                                                                 final Environment environment) {
+                return new QueueListenerWrapper(argumentResolverService, sqsAsyncClient, queueResolverService, environment);
             }
 
             @Bean
             public QueueWrapper coreProvidedPrefetchingQueueListenerWrapper(final ArgumentResolverService argumentResolverService,
                                                                             final SqsAsyncClient sqsAsyncClient,
-                                                                            final QueueResolverService queueResolverService) {
-                return new PrefetchingQueueListenerWrapper(argumentResolverService, sqsAsyncClient, queueResolverService);
+                                                                            final QueueResolverService queueResolverService,
+                                                                            final Environment environment) {
+                return new PrefetchingQueueListenerWrapper(argumentResolverService, sqsAsyncClient, queueResolverService, environment);
             }
 
             @Bean
             public QueueWrapper coreProvidedBatchingQueueListenerWrapper(final ArgumentResolverService argumentResolverService,
                                                                          final SqsAsyncClient sqsAsyncClient,
-                                                                         final QueueResolverService queueResolverService) {
-                return new BatchingQueueListenerWrapper(argumentResolverService, sqsAsyncClient, queueResolverService);
+                                                                         final QueueResolverService queueResolverService,
+                                                                         final Environment environment) {
+                return new BatchingQueueListenerWrapper(argumentResolverService, sqsAsyncClient, queueResolverService, environment);
             }
         }
     }

java-dynamic-sqs-listener-spring/java-dynamic-sqs-listener-spring-starter/src/main/java/com/jashmore/sqs/spring/container/basic/QueueListener.java

@@ -63,11 +63,26 @@
     /**
      * The number of threads that will be processing messages.
      *
+     * <p>This value is ignored when {@link #concurrencyLevelString()} has been set and is not an empty string.
+     *
      * @return the total number of threads processing messages
      * @see ConcurrentMessageBrokerProperties#getConcurrencyLevel() for more details and constraints
      */
     int concurrencyLevel() default 5;
 
+    /**
+     * The number of threads that will be processing messages converted from a string representation.
+     *
+     * <p>This can be used when you need to load the value from Spring properties for example <pre>concurrencyLevelString = "${my.profile.property}"</pre>
+     * instead of having it hardcoded in {@link #concurrencyLevel()}.
+     *
+     * <p>If this value is not empty, the value set by {@link #concurrencyLevel()} will be ignored.
+     *
+     * @return the total number of threads processing messages as a string
+     * @see ConcurrentMessageBrokerProperties#getConcurrencyLevel() for more details and constraints
+     */
+    String concurrencyLevelString() default "";
+
     /**
      * The maximum period of time that the {@link BatchingMessageRetriever} will wait for all threads to be ready before retrieving messages.
      *
@@ -80,11 +95,35 @@
      */
     long maxPeriodBetweenBatchesInMs() default 2000L;
 
+    /**
+     * The maximum period of time that the {@link BatchingMessageRetriever} will wait for all threads to be ready before retrieving messages converted
+     * from a string representation.
+     *
+     * <p>This can be used when you need to load the value from Spring properties for example
+     * <pre>maxPeriodBetweenBatchesInMsString = "${my.profile.property}"</pre> instead of having it hardcoded in {@link #maxPeriodBetweenBatchesInMs()}.
+     *
+     * @return the period in ms that threads will wait for messages to be requested from SQS
+     * @see BatchingMessageRetrieverProperties#getMessageRetrievalPollingPeriodInMs() for more details
+     * @see #maxPeriodBetweenBatchesInMs() for more information about this field
+     */
+    String maxPeriodBetweenBatchesInMsString() default "";
+
     /**
      * The message visibility that will be used for messages obtained from the queue.
      *
      * @return the message visibility for messages fetched from the queue
      * @see BatchingMessageRetrieverProperties#getVisibilityTimeoutInSeconds() for more details and constraints
      */
     int messageVisibilityTimeoutInSeconds() default 30;
+
+    /**
+     * The message visibility that will be used for messages obtained from the queue converted from a string representation.
+     *
+     * <p>This can be used when you need to load the value from Spring properties for example
+     * <pre>messageVisibilityTimeoutInSeconds = "${my.profile.property}"</pre> instead of having it hardcoded in {@link #messageVisibilityTimeoutInSeconds()}.
+     *
+     * @return the message visibility for messages fetched from the queue
+     * @see BatchingMessageRetrieverProperties#getVisibilityTimeoutInSeconds() for more details and constraints
+     */
+    String messageVisibilityTimeoutInSecondsString() default "";
 }

java-dynamic-sqs-listener-spring/java-dynamic-sqs-listener-spring-starter/src/main/java/com/jashmore/sqs/spring/container/basic/QueueListenerWrapper.java

@@ -1,5 +1,7 @@
 package com.jashmore.sqs.spring.container.basic;
 
+import com.google.common.annotations.VisibleForTesting;
+
 import com.jashmore.sqs.QueueProperties;
 import com.jashmore.sqs.argument.ArgumentResolverService;
 import com.jashmore.sqs.broker.concurrent.ConcurrentMessageBroker;
@@ -9,6 +11,7 @@
 import com.jashmore.sqs.processor.MessageProcessor;
 import com.jashmore.sqs.resolver.MessageResolver;
 import com.jashmore.sqs.resolver.individual.IndividualMessageResolver;
+import com.jashmore.sqs.retriever.MessageRetriever;
 import com.jashmore.sqs.retriever.batching.BatchingMessageRetriever;
 import com.jashmore.sqs.retriever.batching.BatchingMessageRetrieverProperties;
 import com.jashmore.sqs.retriever.batching.StaticBatchingMessageRetrieverProperties;
@@ -17,8 +20,9 @@
 import com.jashmore.sqs.spring.QueueWrapper;
 import com.jashmore.sqs.spring.queue.QueueResolverService;
 import com.jashmore.sqs.spring.util.IdentifierUtils;
-import lombok.RequiredArgsConstructor;
+import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.core.env.Environment;
 import org.springframework.util.StringUtils;
 import software.amazon.awssdk.services.sqs.SqsAsyncClient;
 
@@ -29,11 +33,12 @@
  * implementations of the framework.
  */
 @Slf4j
-@RequiredArgsConstructor
+@AllArgsConstructor
 public class QueueListenerWrapper extends AbstractQueueAnnotationWrapper<QueueListener> {
     private final ArgumentResolverService argumentResolverService;
     private final SqsAsyncClient sqsAsyncClient;
     private final QueueResolverService queueResolverService;
+    private final Environment environment;
 
     @Override
     protected Class<QueueListener> getAnnotationClass() {
@@ -47,14 +52,9 @@ protected IdentifiableMessageListenerContainer wrapMethodContainingAnnotation(fi
                 .queueUrl(queueResolverService.resolveQueueUrl(annotation.value()))
                 .build();
 
-        final BatchingMessageRetrieverProperties batchingMessageRetrieverProperties = StaticBatchingMessageRetrieverProperties
-                .builder()
-                .visibilityTimeoutInSeconds(annotation.messageVisibilityTimeoutInSeconds())
-                .messageRetrievalPollingPeriodInMs(annotation.maxPeriodBetweenBatchesInMs())
-                .numberOfThreadsWaitingTrigger(annotation.concurrencyLevel())
-                .build();
-        final BatchingMessageRetriever messageRetriever = new BatchingMessageRetriever(
-                queueProperties, sqsAsyncClient, batchingMessageRetrieverProperties);
+        final int concurrencyLevel = getConcurrencyLevel(annotation);
+
+        final MessageRetriever messageRetriever = buildMessageRetriever(annotation, queueProperties, concurrencyLevel);
 
         final MessageResolver messageResolver = new IndividualMessageResolver(queueProperties, sqsAsyncClient);
 
@@ -73,7 +73,7 @@ protected IdentifiableMessageListenerContainer wrapMethodContainingAnnotation(fi
                 messageProcessor,
                 StaticConcurrentMessageBrokerProperties
                         .builder()
-                        .concurrencyLevel(annotation.concurrencyLevel())
+                        .concurrencyLevel(concurrencyLevel)
                         .threadNameFormat(identifier + "-%d")
                         .build()
         );
@@ -83,4 +83,42 @@ protected IdentifiableMessageListenerContainer wrapMethodContainingAnnotation(fi
                 .container(new SimpleMessageListenerContainer(messageRetriever, messageBroker, messageResolver))
                 .build();
     }
+
+    private MessageRetriever buildMessageRetriever(final QueueListener annotation, final QueueProperties queueProperties, final int concurrencyLevel) {
+        return new BatchingMessageRetriever(
+                queueProperties, sqsAsyncClient, batchingMessageRetrieverProperties(annotation, concurrencyLevel));
+    }
+
+    @VisibleForTesting
+    BatchingMessageRetrieverProperties batchingMessageRetrieverProperties(final QueueListener annotation, final int concurrencyLevel) {
+        return StaticBatchingMessageRetrieverProperties.builder()
+                .visibilityTimeoutInSeconds(getMessageVisibilityTimeoutInSeconds(annotation))
+                .messageRetrievalPollingPeriodInMs(getMaxPeriodBetweenBatchesInMs(annotation))
+                .numberOfThreadsWaitingTrigger(concurrencyLevel)
+                .build();
+    }
+
+    private int getConcurrencyLevel(final QueueListener annotation) {
+        if (StringUtils.isEmpty(annotation.concurrencyLevelString())) {
+            return annotation.concurrencyLevel();
+        }
+
+        return Integer.parseInt(environment.resolvePlaceholders(annotation.concurrencyLevelString()));
+    }
+
+    private long getMaxPeriodBetweenBatchesInMs(final QueueListener annotation) {
+        if (StringUtils.isEmpty(annotation.maxPeriodBetweenBatchesInMsString())) {
+            return annotation.maxPeriodBetweenBatchesInMs();
+        }
+
+        return Long.parseLong(environment.resolvePlaceholders(annotation.maxPeriodBetweenBatchesInMsString()));
+    }
+
+    private int getMessageVisibilityTimeoutInSeconds(final QueueListener annotation) {
+        if (StringUtils.isEmpty(annotation.messageVisibilityTimeoutInSecondsString())) {
+            return annotation.messageVisibilityTimeoutInSeconds();
+        }
+
+        return Integer.parseInt(environment.resolvePlaceholders(annotation.messageVisibilityTimeoutInSecondsString()));
+    }
 }

java-dynamic-sqs-listener-spring/java-dynamic-sqs-listener-spring-starter/src/main/java/com/jashmore/sqs/spring/container/batching/BatchingQueueListener.java

@@ -72,6 +72,19 @@
      */
     int concurrencyLevel() default 5;
 
+    /**
+     * The number of threads that will be processing messages converted from a string representation.
+     *
+     * <p>This can be used when you need to load the value from Spring properties for example <pre>concurrencyLevelString = "${my.profile.property}"</pre>
+     * instead of having it hardcoded in {@link #concurrencyLevel()}.
+     *
+     * <p>If this value is not empty, the value set by {@link #concurrencyLevel()} will be ignored.
+     *
+     * @return the total number of threads processing messages as a string
+     * @see ConcurrentMessageBrokerProperties#getConcurrencyLevel() for more details and constraints
+     */
+    String concurrencyLevelString() default "";
+
     /**
      * The maximum period of time that the {@link BatchingMessageRetriever} will wait for all threads to be ready before retrieving messages.
      *
@@ -84,11 +97,35 @@
      */
     long maxPeriodBetweenBatchesInMs() default 2000L;
 
+    /**
+     * The maximum period of time that the {@link BatchingMessageRetriever} will wait for all threads to be ready before retrieving messages converted
+     * from a string representation.
+     *
+     * <p>This can be used when you need to load the value from Spring properties for example
+     * <pre>maxPeriodBetweenBatchesInMsString = "${my.profile.property}"</pre> instead of having it hardcoded in {@link #maxPeriodBetweenBatchesInMs()}.
+     *
+     * @return the period in ms that threads will wait for messages to be requested from SQS
+     * @see BatchingMessageRetrieverProperties#getMessageRetrievalPollingPeriodInMs() for more details
+     * @see #maxPeriodBetweenBatchesInMs() for more information about this field
+     */
+    String maxPeriodBetweenBatchesInMsString() default "";
+
     /**
      * The message visibility that will be used for messages obtained from the queue.
      *
      * @return the message visibility for messages fetched from the queue
      * @see BatchingMessageRetrieverProperties#getVisibilityTimeoutInSeconds() for more details and constraints
      */
     int messageVisibilityTimeoutInSeconds() default 30;
+
+    /**
+     * The message visibility that will be used for messages obtained from the queue converted from a string representation.
+     *
+     * <p>This can be used when you need to load the value from Spring properties for example
+     * <pre>messageVisibilityTimeoutInSeconds = "${my.profile.property}"</pre> instead of having it hardcoded in {@link #messageVisibilityTimeoutInSeconds()}.
+     *
+     * @return the message visibility for messages fetched from the queue
+     * @see BatchingMessageRetrieverProperties#getVisibilityTimeoutInSeconds() for more details and constraints
+     */
+    String messageVisibilityTimeoutInSecondsString() default "";
 }