refs #77: Made AsyncMessageRetriever a Runnable instead of using start/stop methods.

Author: jashmoreAtlassian

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

@@ -16,6 +16,8 @@
 import com.jashmore.sqs.broker.concurrent.ConcurrentMessageBroker;
 import com.jashmore.sqs.broker.concurrent.properties.CachingConcurrentMessageBrokerProperties;
 import com.jashmore.sqs.broker.concurrent.properties.ConcurrentMessageBrokerProperties;
+import com.jashmore.sqs.container.MessageListenerContainer;
+import com.jashmore.sqs.container.SimpleMessageListenerContainer;
 import com.jashmore.sqs.processor.DefaultMessageProcessor;
 import com.jashmore.sqs.processor.MessageProcessor;
 import com.jashmore.sqs.resolver.AsyncMessageResolver;
@@ -95,13 +97,9 @@ public static void main(final String[] args) throws Exception {
                         .desiredMinPrefetchedMessages(10)
                         .maxPrefetchedMessages(20)
                         .maxWaitTimeInSecondsToObtainMessagesFromServer(10)
-                        .build(),
-                executorService
+                        .build()
         );
 
-        // As this retrieves messages asynchronously we need to start the background thread
-        messageRetriever.start();
-
         // 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);
@@ -139,11 +137,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();
+        final MessageListenerContainer messageListenerContainer
+                = new SimpleMessageListenerContainer(messageRetriever, concurrentMessageBroker, messageResolver);
+        messageListenerContainer.start();
 
         // Create some producers of messages
         final Future<?> producerFuture = executorService.submit(new Producer(sqsAsyncClient, queueUrl));

examples/java-dynamic-sqs-listener-spring-starter-examples/src/main/java/com/jashmore/sqs/examples/MessageListeners.java

@@ -14,7 +14,7 @@ public class MessageListeners {
      *
      * @param payload the payload of the SQS Message
      */
-    @QueueListener(value = "test")
+    @QueueListener(value = "test", identifier = "test")
     public void method(@Payload final String payload) {
         log.info("Message Received: {}", payload);
     }

examples/java-dynamic-sqs-listener-spring-starter-examples/src/main/java/com/jashmore/sqs/examples/ScheduledQueueListenerEnabler.java

@@ -0,0 +1,32 @@
+package com.jashmore.sqs.examples;
+
+import com.jashmore.sqs.spring.QueueContainerService;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class ScheduledQueueListenerEnabler {
+    // TODO: #79: Allow for dependency injection of QueueContainerService
+    private final BeanFactory beanFactory;
+
+    /**
+     * Just a scheduled job that shows that there are methods of turning off the container when necessary.
+     *
+     * @throws InterruptedException if the thread was interrupted while sleeping
+     */
+    @Scheduled(initialDelay = 10_000, fixedDelay = 30_000)
+    public void turnOfSqsListener() throws InterruptedException {
+        log.info("Turning off SQS Listener for a short period");
+
+        final QueueContainerService queueContainerService = beanFactory.getBean(QueueContainerService.class);
+        queueContainerService.stopContainer("test");
+        Thread.sleep(5_000);
+        log.info("Turning SQS Listener back om");
+        queueContainerService.startContainer("test");
+    }
+}

java-dynamic-sqs-listener-api/src/main/java/com/jashmore/sqs/retriever/AsyncMessageRetriever.java

@@ -1,43 +1,25 @@
 package com.jashmore.sqs.retriever;
 
-import java.util.concurrent.Future;
-
 /**
  * {@link MessageRetriever} that obtains messages in the background for consumption.
  *
  * <p>For example, this could be used when messages are pulled from the queue and cached locally to improve performance by reducing the number
  * of times and latency of pulling messages from the queue.
+ *
+ * <p>To utilise this retriever, the class must run on a separate thread. For example:
+ *
+ * <pre class="code">
+ *     final AsyncMessageRetriever messageRetriever = new SomeAsyncMessageRetrieverImpl(...);
+ *     // start it on a background thread
+ *     Future&lt;?&gt; retrieverFuture = Executors.newCachedThreadPool().submit(messageRetriever);
+ *
+ *     // Now messages can be retrieved
+ *     Message message = retrieverFuture.retrieveMessage(message);
+ *
+ *     // Stop the message retriever when you are done
+ *     retrieverFuture.cancel(true);
+ * </pre>
  */
-public interface AsyncMessageRetriever extends MessageRetriever {
-    /**
-     * Start retrieve messages from the queue in a background thread.
-     *
-     * <p>Requirements for this method include:
-     *
-     * <ul>
-     *     <li>This method must be non-blocking and return once the background thread has started.</li>
-     *     <li>If this retriever has already been started, any calls to this method will throw an {@link IllegalStateException}.</li>
-     *     <li>If this retriever is being stopped by calling {@link #stop()}, a call to this method should <b>not</b> be blocked by the previous thread and
-     *         should start a new thread.</li>
-     * </ul>
-     *
-     * @throws IllegalStateException if the retriever has already been started
-     */
-    void start();
+public interface AsyncMessageRetriever extends MessageRetriever, Runnable {
 
-    /**
-     * Stop the retrieval of messages.
-     *
-     * <p>Requirements for this method include:
-     *
-     * <ul>
-     *     <li>This method must be non-blocking and return once the background thread has been triggered to stop.</li>
-     *     <li>If this retriever has not been started or has already been stopped, any calls to this method will return a resolved {@link Future}</li>
-     *     <li>The returned {@link Future} does not have any requirements for the value resolved and therefore should not be relied upon.</li>
-     * </ul>
-     *
-     * @return future that will resolve when the background message retriever thread has stopped
-     * @throws IllegalStateException if the retriever has not been started or has already stopped
-     */
-    Future<Object> stop();
 }

java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/container/SimpleMessageListenerContainer.java

@@ -8,12 +8,14 @@
 import com.jashmore.sqs.resolver.MessageResolver;
 import com.jashmore.sqs.retriever.AsyncMessageRetriever;
 import com.jashmore.sqs.retriever.MessageRetriever;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import lombok.extern.slf4j.Slf4j;
 
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 import javax.annotation.concurrent.GuardedBy;
 
 /**
@@ -45,18 +47,8 @@ public class SimpleMessageListenerContainer implements MessageListenerContainer
      */
     private final MessageResolver messageResolver;
 
-    private final ExecutorService executorService;
-
-    /**
-     * Stores whether the container is currently running.
-     *
-     * <p>This is kept thread safe by making sure all methods for this container are synchronized.
-     */
-    @GuardedBy("this")
-    private volatile boolean isRunning;
-
     @GuardedBy("this")
-    private Future<?> messageResolverCompletableFuture;
+    private ExecutorService executorService;
 
     /**
      * Container that can be built when the {@link MessageBroker} is using an {@link AsyncMessageRetriever}. This takes the {@link AsyncMessageRetriever} so
@@ -72,64 +64,62 @@ public SimpleMessageListenerContainer(final MessageRetriever messageRetriever,
         this.messageRetriever = messageRetriever;
         this.messageBroker = messageBroker;
         this.messageResolver = messageResolver;
-        this.executorService = Executors.newCachedThreadPool();
 
-        this.messageResolverCompletableFuture = null;
-    }
-
-    @VisibleForTesting
-    SimpleMessageListenerContainer(final MessageRetriever messageRetriever,
-                                   final MessageBroker messageBroker,
-                                   final MessageResolver messageResolver,
-                                   final ExecutorService executorService) {
-        this.messageRetriever = messageRetriever;
-        this.messageBroker = messageBroker;
-        this.messageResolver = messageResolver;
-        this.executorService = executorService;
-
-        this.messageResolverCompletableFuture = null;
+        this.executorService = null;
     }
 
+    @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED_BAD_PRACTICE")
     @Override
     public synchronized void start() {
-        if (isRunning) {
+        if (executorService != null) {
             return;
         }
 
+        executorService = getNewExecutorService();
+
         if (messageRetriever instanceof AsyncMessageRetriever) {
-            ((AsyncMessageRetriever)messageRetriever).start();
+            executorService.submit((AsyncMessageRetriever) messageRetriever);
         }
 
         if (messageResolver instanceof AsyncMessageResolver) {
-            messageResolverCompletableFuture = executorService.submit((AsyncMessageResolver) messageResolver);
+            executorService.submit((AsyncMessageResolver) messageResolver);
         }
 
         messageBroker.start();
-
-        isRunning = true;
     }
 
     @Override
     public synchronized void stop() {
-        if (!isRunning) {
+        if (executorService == null) {
             return;
         }
 
         try {
-            final Future<?> messageBrokerStoppedFuture = messageBroker.stop();
-            if (messageRetriever instanceof AsyncMessageRetriever) {
-                ((AsyncMessageRetriever)messageRetriever).stop().get();
-            }
+            executorService.shutdownNow();
 
-            if (messageResolverCompletableFuture != null) {
-                messageResolverCompletableFuture.cancel(true);
+            try {
+                final Future<?> messageBrokerStoppedFuture = messageBroker.stop();
+
+                messageBrokerStoppedFuture.get();
+            } catch (final InterruptedException interruptedException) {
+                Thread.currentThread().interrupt();
+            } catch (final ExecutionException executionException) {
+                log.error("Error waiting for container to stop", executionException.getCause());
             }
 
-            messageBrokerStoppedFuture.get();
-        } catch (final InterruptedException | ExecutionException exception) {
-            log.error("Error waiting for container to stop", exception.getCause());
+            try {
+                executorService.awaitTermination(1, TimeUnit.MINUTES);
+            } catch (final InterruptedException interruptedException) {
+                Thread.currentThread().interrupt();
+            }
         } finally {
-            isRunning = false;
+            // Reset so we can start the container again in the future
+            executorService = null;
         }
     }
+
+    @VisibleForTesting
+    ExecutorService getNewExecutorService() {
+        return Executors.newCachedThreadPool();
+    }
 }