Merge pull request #75 from JaidenAshmore/issue/66_message_resolver_promotion

Issue/66 message resolver promotion

Author: JaidenAshmore

README.md

@@ -1,5 +1,5 @@
 # Java Dynamic SQS Listener
-[![Build Status](https://semaphoreci.com/api/v1/jaidenashmore/java-dynamic-sqs-listener/branches/master/badge.svg)](https://semaphoreci.com/jaidenashmore/java-dynamic-sqs-listener)
+[![Build Status](https://semaphoreci.com/api/v1/jaidenashmore/java-dynamic-sqs-listener/branches/master/shields_badge.svg)](https://semaphoreci.com/jaidenashmore/java-dynamic-sqs-listener)
 [![Coverage Status](https://coveralls.io/repos/github/JaidenAshmore/java-dynamic-sqs-listener/badge.svg?branch=master)](https://coveralls.io/github/JaidenAshmore/java-dynamic-sqs-listener?branch=master)
 [![Total alerts](https://img.shields.io/lgtm/alerts/g/JaidenAshmore/java-dynamic-sqs-listener.svg?logo=lgtm&logoWidth=18)](https://lgtm.com/projects/g/JaidenAshmore/java-dynamic-sqs-listener/alerts/)
 [![Language grade: Java](https://img.shields.io/lgtm/grade/java/g/JaidenAshmore/java-dynamic-sqs-listener.svg?logo=lgtm&logoWidth=18)](https://lgtm.com/projects/g/JaidenAshmore/java-dynamic-sqs-listener/context:java)
@@ -30,10 +30,12 @@ the processing of a message from the queue by delegating it to the corresponding
 arguments of the method being executed from the message. For example, a parameter with the
 [@Payload](./java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/argument/payload/Payload.java) annotation will be resolved with the
 body of the message cast to that type (e.g. a POJO).
-- The [MessageBroker](./java-dynamic-sqs-listener-api/src/main/java/com/jashmore/sqs/broker) is the main container that controls the whole flow
+- The [MessageBroker](./java-dynamic-sqs-listener-api/src/main/java/com/jashmore/sqs/broker/MessageBroker.java) is the main container that controls the whole flow
 of messages from the [MessageRetriever](./java-dynamic-sqs-listener-api/src/main/java/com/jashmore/sqs/retriever/MessageRetriever.java) to the
 [MessageProcessor](./java-dynamic-sqs-listener-api/src/main/java/com/jashmore/sqs/processor/MessageProcessor.java). It can determine when more messages
 are to be processed and the rate of concurrency for processing messages.
+- The [MessageResolver](./java-dynamic-sqs-listener-api/src/main/java/com/jashmore/sqs/resolver/MessageResolver.java) is used when the message has been
+successfully processed and it is needed to be removed from the SQS queue so it isn't processed again.
 
 See the [Core Implementations Overview](./doc/core-implementations-overview.md) for more information about the core implementations provided by this library.
 

doc/core-implementations-overview.md

@@ -77,7 +77,7 @@ will be injected with an implementation that extends the message visibility of t
 [VisibilityExtenderArgumentResolver](../java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/argument/visibility/VisibilityExtenderArgumentResolver.java). 
 
 ### Message Broker
-The [MessageBroker](../java-dynamic-sqs-listener-api/src/main/java/com/jashmore/sqs/broker) is the main container that controls the whole flow
+The [MessageBroker](../java-dynamic-sqs-listener-api/src/main/java/com/jashmore/sqs/broker/MessageBroker.java) is the main container that controls the whole flow
 of messages from the [MessageRetriever](../java-dynamic-sqs-listener-api/src/main/java/com/jashmore/sqs/retriever) to the
 [MessageProcessor](../java-dynamic-sqs-listener-api/src/main/java/com/jashmore/sqs/processor/MessageProcessor.java). It can provide logic like the rate
 of concurrency of the messages being processed or when messages should be processed.
@@ -88,4 +88,23 @@ this implementation only runs on a single thread and therefore only a single mes
 for local development and testing and does not have a significant production use case.
 - [ConcurrentMessageBroker](../java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/broker/concurrent/ConcurrentMessageBroker.java): this
 implementation will run on multiple threads each processing messages. It has dynamic configuration and this allows the rate of concurrency to change
-dynamically while the application is running.
+dynamically while the application is running.
+
+### Message Resolver
+The [MessageResolver](../java-dynamic-sqs-listener-api/src/main/java/com/jashmore/sqs/resolver/MessageResolver.java) is used when the message has been
+successfully processed and it is needed to be removed from the SQS queue so it isn't processed again.
+
+Core implementation include:
+- [IndividualMessageResolver](../java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/resolver/individual/IndividualMessageResolver.java): this
+implementation will immediately call out to SQS to remove the message from the queue asynchronously. This is useful for when you want the message to be
+immediately removed and the extra calls to SQS is not a problem when there are many messages being deleted at once.
+- [BatchingMessageResolver](../java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/resolver/batching/BatchingMessageResolver.java): this
+implementation will batch calls to delete messages from the SQS queue into a batch that will go out together once asynchronously. This is useful if you
+are processing many messages at the same time and it is desirable to reduce the number of calls out to SQS. A disadvantage is that the message may
+sit in the batch for enough time that the visibility expires and it is placed onto the queue. To mitigate this, smaller batch
+timeout should be used or by increasing the visibility timeout. 
+- [BlockingMessageResolver](../java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/resolver/blocking/BlockingMessageResolver.java): this
+implementation will delegate to another [MessageResolver](../java-dynamic-sqs-listener-api/src/main/java/com/jashmore/sqs/resolver/MessageResolver.java)
+implementation and will block the thread until the underlying implementation finishes resolving the message. This can be useful if you don't want to
+process any more messages  on this thread until the previous is successfully removed from SQS. It is also useful for testing as it reduces the amount of
+asynchronous code running and therefore can reduce flaky tests.

doc/documentation.md

@@ -19,8 +19,6 @@ 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 add use the custom queue listener](how-to-guides/spring/spring-how-to-use-custom-queue-listener.md): useful for learning how to use custom
-        components of the framework in a queue listener without creating your own annotation
         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-use-custom-queue-listener.md

@@ -1 +1 @@
-<mxfile modified="2018-12-31T04:57:38.346Z" host="www.draw.io" agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_2) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/70.0.3538.110 Safari/537.36" etag="l177sfoR97Yz4h89gFaG" version="10.0.13" type="device"><diagram name="Page-1" id="55a83fd1-7818-8e21-69c5-c3457e3827bb">5Vpbd6LKEv41rnX2w8ziaswjAknIsVEDGaNvCAQb0XYDyuXX72ouXiAmGSeXvc6ZNYl0dXV11VdfVzdtOry8Sm9Da7NAxHGDDsc4aYdXOhx3LTLwmwqyUiCK16XAC7FTitiDwMC5Wwmrcd4WO250ohgTEsR4cyq0yXrt2vGJzApDkpyqPZPgdNaN5bktgWFbQVs6wU68qKQswxw67lzsLaqpe3XAc8teeiHZrqv5Ohz/XPwru1dWbavSjxaWQ5IjEa92eDkkJC6fVqnsBhTaGrZy3M2Z3r3fobuO3zOg312MohH673Bh3C0iayo8+6sflZWdFWzdOozC2TirAQK/N/QRrwok+8WnFG3KZDAgserGM05dmK2/c8MYA8QDa+4GIxLhGJM19M9JHJPVkYIUYI92xGRTW4aWY8VWh5fKJncT7bwO109XEIw8utO5WdYX5pN0a+cMtu4eGFshuwHv8E4m8igTd/bK3iFfSpB8nTsrG2t3zmZ290BGhpYhxU413CezSbC27sbXmq+xSJFAZ7aZPTnynPdAJnlIljLdnHool0TQT+zVr3z2dM/NHm9y+/aX79zcL5zbYDcy7gVzEiwnj+xmJmvg5+iuDz2eN1MYbJoaqxuCODRVD5lSMvC1FE2SWJeFFJlaPPAlbmgwKfLRVs+X2cBE3MCfsgNDSPR8HCND4IbmbILUpPBG98db8Jabywz9FDWF6muJbmjebBVEc5hTY8bi2LQZZGi7kZ8m06cHot1CpEs6jwc2Hrf6JIHxaj4wYS7/FwI/Uj2TOJSBr4paRq+Mt0NzCXOBL8W8KHOwkCDlPgLf+YG/zFA1Rs/Bf0WjfrMQK8SlcrrBcEgWBD2nbUDZAJ9zJJqKmpX21a1uSBmiP+Z9NPAfY2SqDHxSrLiinWtYu51t5rfJtYYR2JqKyF/iQS707NsbxpL7S8iAjswx+KNCzFI6NB+3oCMARilkj+IK9jyYeylCnEmBN7Rhfl4DDMBHfkjHyQKD/HFc6kgCKmIZH82vq8NHlYdxrfmHirQdKlMG7Al6JgjI1CNdsbfIRJmmTL3CpszwMAenm15c9aWa8kgxLvKNlCXkZrqFn9w0UT7wIV7qF8Upt9O5goA/j2Bfgj5PAFt0TKz7GsQ2hVwW/aKOi35R56cs8IzVc++dGE6pHa/EbwE5VlPKRZQxOeSYGSpjr8o7xCHlBV4mzZkkDHybo1jC/AlaJqmOhRz8EIH72VA+4mY+zXRVTR/a3AR+Uyw0mhMeMMz1fBYNTDWDZxHldP2Af+aYPcIyB6y8Iy7SeTNYL5xe4Kqx9fqDfHtQEcBPL9dvkxhhgddNBPZUsZgDOAC4wGq3KdbAASTSmIE7NR844IOA6Noo58sR5TflvnnDletWS09ifWUdFvjQmuBTzKaiziXAScizP2ZhTRc40twj36Zx0LoAviNuSDmteNRHHvymeoBzm+dQsVbWJI2g4vkPyxsV+qDq9Wh1KmsiPEG9rWo/1GI3PbupsPutCk4ALlm5cZiBSjWgVw+pdn+hV+12yWEvFXulaHG0i15Vala1e3t7y4cdDB6qTew3NjSutaEhN4pgI3kA89iFYFsbXLLAsWtsLJu2EzjkADSLGLYbXmE/BiWx9zZKbH1O+BKY+HMwjUJiwxP5Dph44d8Gk9CCSQq97QrieXAjEkC0hhvuMEDCdQNwpj8H3LpevMfja/FrLMYu8wJ+3a/ETzxHs35Ilt+yFIXuKUb81XdzrPvCEbwkk4N3J/B0/97Sl4U+RB//sMqjswQagfscH3oPBCytRBtrXct2BFyBd5MMufGCOP+pdegp+0jtSFz4cCr9eLfg01rRXK/n0aZoM5eLBGZkZQGxaKAoGxGf0Igr2hF4TeXk/82wq6Wl0cAN2OzW3iFwKnw15mPZX9TeVf97ULoo/JviP1NtXoewizUcul+e2CvlN8H78jrYPLd1r36KrULIvVQIr8VPKoRXrULI/4R26Nou3rmozmgDKnftSPQaCFp2YEURtk+xAjzC7Km6qSgaU9qAaKGZ4vip1oPnQ085jeu07o4a+IIrZBva7tsn0tgKPTd+6yqmna/jg/QL2ahloRtYMcB0ej/2QoqqGUYEQyB7Ovxgr0/4cNVtbHdlmNWo41umhiH+lFdXbMNOCUPLTkGYfdSXc6jX4hD7s+CLFcZ/QJ2SLXvy/BZB3ky8+J2JFxuFgG8m7L2JF5mGIe5rM8++4yrzTxJe1wrmM2uF+E7K8N9JGb7xNsv3LqXMZbUCUmhlR2obqhCd91doMFNgxAYBS4sfS8f2G7ZAK9GmPKKc283omeOUk6Eb4dyaFwqUe1W0oC32O6JCr9+rS3Qb6AbvVO3b9RV2HDq+H9D7+P7+ewuZBPCeT+etv7l45WhRff9SedLZHw2Oaf7Kwjx7EGF+Cs3rkfpbiUt5WquQ5+fI/Zxi075m4sqzSnnJ9OeHlfMF6FMPK+8tQNxbORV61QL4/zi//G5N6javul6rSV/K7PaVl1gyu7jvqq+/Pmdr3R/Kp53jF50PYTb/TmYL37m1Nsh3KYkv3Fg/jEPta78u5RBe78jyD4oi2zn7Bvepp7L3Uqf7rdRpVJTupacy9rphiP8o9kDz8OcOpfrhT0p49R8=</diagram></mxfile>
+<mxfile modified="2019-04-23T00:16:36.657Z" host="www.draw.io" agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/73.0.3683.103 Safari/537.36" etag="1lyKOm5FsYB_9nP8qhof" version="10.6.3" type="device"><diagram name="Page-1" id="55a83fd1-7818-8e21-69c5-c3457e3827bb">5Vprd6JIE/41nrPvh8nh1mo+opCErI0ayRj9hoAIovgCyuXXbzUXRVBj3CQzuztn5khXN9VVT13pngbbXUWPnrpZYFc3nAZD6VGDFRoMQ7P3HPwQSpxRWs2cYHqWni86EEZWYuREKqduLd3wjxYGrusE1uaYqLnrtaEFRzTV89zweNncdY533aimUSOMNNWpU8eWHixyKk1Rh4knwzIX+dZtlE/MVG1peu52ne/XYNh5+iebXqkFr3y9v1B1NyyRWLHBdj3XDbKnVdQ1HIJtAVv23sOZ2b3cnrEOrnmh01wM/AH+s78YPS18dcLN7dWPnMtOdbZGoUYqbBAXAIHcG/JorVIkO+kv728yY1BAUYvB3IoM2K2zM7zAAoh76sxwBq5vBZa7hvmZGwTuqrSAdyyTTATupuAMI10N1AbLZ0Pmwd+ZDaYTrUCZ7uBJZqZxh5uNo62WUJb69EJpgrvrsTqrx4jFMdppK22HbT7E3ftEX2mW9KRvpk8v7mAkxVjQIsnquNOxs1afhveSLdFY4GHNdDN907sz1gQab+IuH8vKxMQJj2B9qK1+JtO3Z2b6+pBojz9t/eF5oT86u8HomVPGznL8Sm+mXQnkHDx1YMY0pwJlKYpEyyMO9RXRxAof9mwpwuMwkLtchBUp6Nk80x9REbbxVk6WcU/BTM+e0L0RF8rJMMAjjukr0zEWw1Qa2R5uQVpm1qXIL5IEsl4K5ZFkTleOP4M9JWqIhopG4ZG0G9hROHl7caVH0HRJ9jGBx+tWHofwvpj0FNjL/olBjkiOeQbHIKsgZtoLw21fWcJeIEu6L451iwux8OyD7GzPXsY4f0dOQH5BInLToCvoJTLyiGJwl+PkhIwB5RHInGCkCGKc8Re38oiPMfmnPPs9+zXAikjBL8GKSceJZEmP083sMbyXLAy8JgjbS6uXcG3t8YFSu50lWEDGyhDkEUFnPuorr1tYwwFGEViP4Ar8TNh7iUDPMMUbxrA/KwEGICPbJ+91OQrbwyBbw3M41WVY2l8W+68iC+/V9u8L/LYvTCjgx8kxx2FF9mVB22IFx5IwMVOeXYqFPRhZMYN8LpKEV4Jxam8sLME2ky38SxQFJz0b9CVyEZwSLZoJGPznFfjzMGdywIu8E8i2BLpNwJbpPJKtdB7J7IQGP6PlxLwSwwnhY2b4LcDGYkR8EcdUAjam+sLQzO0OevBJipdCbMZzPVtjCJawf4iXYSRbXAJyIPD9uN8t+WYyiWVRjF7qvgn+TbCQiE1YwDCRk6nfU8QYnhFOSPyAfMqQLmGZAFZmyRfJvjHECyOnuEp0EX9gbxMyAshpJvJjGGCLY2UFAz8RpXuADwAuEO0awRp8ACOiM/hO4Q8M+AOHSWxk+yWY+DfxfeWByeJWio50vRCHKT4kJ9gEswmSmRB8EuxsD2mI6RRHYntsa0QPkhdAdsz0iU8LJpGRBbnJOsC57ueQsVbqOPIh49kvywcR5iDrtUl2ynIiPEG+rVeQohxAejaiEimvKI+GuzICL4Yl+WybzV/Jqz8qynp4qKWonZEWpSraypepefU295wPFQwe8iL2gYLG1AoaNnwfCskLsLcM0KxW4MKFFRijjaqRcQhdDkCzCKDcsAL9SSgVzUSOEtuqo0QXfcJnw7ScK02sjsatznw+Ml5Wq+en5gmYWncw9gzfdXZGDlkNKUOH1ikful6wcE13rTrigdpJ+yHSA6SNwWFNzyUVPsXTNoIgzvtAdRu4x2gbkRW8lZ4nhNUdykdClHNOB3E+yOQkwp1tjXKS7249zbjgPmzegaqeaQQX8Lts7x/UHYWaRyanUTb0DEcNrN2xpKfMmzMfuBbocFjizuc+SFa1/16G2yOHPRc5A8/V4Mn9FZHDcu/nly+LnJMwcTWYeM/crkDFlyx4vJHh7SyAhGk6IExnBrg1zWCPx7fiV/HCk/A1vxM+dM7LIHUsf0lyRhS6Q78uPZ9EqXnisyzzJt3aHQHU/P+WfEB2QNngh5p9TvGwwjHmwWH24IEZF3+jrgvazgVR4Hs1xgZka/2PYg358iotK5FTGY6pny8W/KorYu31zN+kY+p2EkcN1NhxVaIojgeu7RKNc8dzdTBY99+pdh5cElF8BA3Q2jwoTogXdS7T/kf4tTq/BqWb1H9I/1J59TqoncawZ3y7YVvCB8H79kzI3B8X2yZXz4PMqTx4j74oD7ZqeZDN2lTNsC60qWudJyeDMNIc1fctrdJjrkG2t0NjSYaTUmNZmso70KIbPd1/pgwOU+nouDe9pRVl6q3opSO8SvtbN3L5i+yEDQvarQ1q0frS90c+1GpWamSmd/5W+biywqjyYdmiK3wyXGp8Pqsbbtc8j75LvUz1gk91uLMO8q7h0W9leNQ8bjVZ6kbDI+rY8izzvZanrzgT/5QMU/vCvT1XoCtdhv2tXIZtVyzdvtVlbssVYEI1Li3bkAX+eXm5imdyFKo4YMbxc92x/l3OkUy0yfqaczWQNCrHPukZvpWos3QB8b1cW1iNOg0kkHuc/DZGA/+DT7H6Nc3K0vX0mMchFzud/QVY13VcL923uAI75cyXo63apexv/HKRG+Vbs1PdC3XHoYpDFbXx9z10oesHcUzW4WSnlX+/xTlKP/um5swR25c0MtcmJ+bK5LQ3bs74v9HYfDRZNbkPJKtvdfn6CRoqnT0Xp2kfd/mzdfWdzn0fC5Uw+ZJgOHHAfOmY8d1gKJ04M1wl+X1rfFS8+tbo2ItTfIOyXxMeTPt4H8T9LuFRPyFtkvCw1jt3+W/91r02KppXRsU3uXwlxTZv7V/p6sHLlU5/g8+dv806fWmaXWnU3O6qw6gTPV3NOc4fy7evuLr41jvTeiPeJoGpG44R/NOuTC/2h++G+EevWi853XXnWxe6/mrd+7s3rcXHXqty84gqh5zXBnerUtFQ1UHPBPfnFB0YHv6LYbb88P84WfEv</diagram></mxfile>

doc/resources/architecture_diagram.png

@@ -18,12 +18,13 @@
 import com.jashmore.sqs.broker.concurrent.properties.ConcurrentMessageBrokerProperties;
 import com.jashmore.sqs.processor.DefaultMessageProcessor;
 import com.jashmore.sqs.processor.MessageProcessor;
-import com.jashmore.sqs.processor.resolver.MessageResolver;
-import com.jashmore.sqs.processor.resolver.batching.BatchingMessageResolver;
-import com.jashmore.sqs.processor.resolver.batching.StaticBatchingMessageResolverProperties;
+import com.jashmore.sqs.resolver.AsyncMessageResolver;
+import com.jashmore.sqs.resolver.batching.BatchingMessageResolver;
+import com.jashmore.sqs.resolver.batching.StaticBatchingMessageResolverProperties;
 import com.jashmore.sqs.retriever.AsyncMessageRetriever;
 import com.jashmore.sqs.retriever.prefetch.PrefetchingMessageRetriever;
 import com.jashmore.sqs.retriever.prefetch.StaticPrefetchingMessageRetrieverProperties;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
@@ -73,6 +74,7 @@ public class ConcurrentBrokerExample {
      * @param args unused args
      * @throws Exception if there was a problem running the program
      */
+    @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED_BAD_PRACTICE")
     public static void main(final String[] args) throws Exception {
         // Sets up the SQS that will be used
         final SqsAsyncClient sqsAsyncClient = startElasticMqServer();
@@ -103,10 +105,11 @@ public static void main(final String[] args) throws Exception {
         // Creates the class that will deal with taking messages and getting them processed by the message consumer
         final MessageConsumer messageConsumer = new MessageConsumer();
         final Method messageReceivedMethod = MessageConsumer.class.getMethod("method", Request.class, String.class);
-        final MessageResolver messageResolver = new BatchingMessageResolver(queueProperties, sqsAsyncClient, StaticBatchingMessageResolverProperties.builder()
-                .bufferingSizeLimit(MAX_NUMBER_OF_MESSAGES_IN_BATCH)
-                .bufferingTimeInMs(5000)
-                .build());
+        final AsyncMessageResolver messageResolver = new BatchingMessageResolver(queueProperties, sqsAsyncClient,
+                StaticBatchingMessageResolverProperties.builder()
+                        .bufferingSizeLimit(MAX_NUMBER_OF_MESSAGES_IN_BATCH)
+                        .bufferingTimeInMs(5000)
+                        .build());
         final MessageProcessor messageProcessor = new DefaultMessageProcessor(
                 argumentResolverService(sqsAsyncClient),
                 queueProperties,
@@ -136,6 +139,9 @@ public Integer getConcurrencyLevel() {
                 })
         );
 
+        // As the BatchingMessageResolver uses a background thread to delete the messages in batches we need to start it in a background thread
+        executorService.submit(messageResolver);
+
         // When we start listening it will receive messages from SQS and pass them to the MessageConsumer for processing
         concurrentMessageBroker.start();