Merge pull request #81 from JaidenAshmore/issue/77_message_broker_as_runnable

refs 77: Changed message broker to be a runnable

Author: JaidenAshmore

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

@@ -13,9 +13,9 @@
 import com.jashmore.sqs.argument.payload.Payload;
 import com.jashmore.sqs.argument.payload.mapper.JacksonPayloadMapper;
 import com.jashmore.sqs.argument.payload.mapper.PayloadMapper;
+import com.jashmore.sqs.broker.concurrent.CachingConcurrentMessageBrokerProperties;
 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.broker.concurrent.ConcurrentMessageBrokerProperties;
 import com.jashmore.sqs.container.MessageListenerContainer;
 import com.jashmore.sqs.container.SimpleMessageListenerContainer;
 import com.jashmore.sqs.processor.DefaultMessageProcessor;
@@ -120,7 +120,6 @@ public static void main(final String[] args) throws Exception {
         final ConcurrentMessageBroker concurrentMessageBroker = new ConcurrentMessageBroker(
                 messageRetriever,
                 messageProcessor,
-                executorService,
                 // Represents a concurrent implementation that will fluctuate between 0 and 10 threads all processing messages
                 new CachingConcurrentMessageBrokerProperties(10000, new ConcurrentMessageBrokerProperties() {
                     private final Random random = new Random(1);

java-dynamic-sqs-listener-api/src/main/java/com/jashmore/sqs/broker/MessageBroker.java

@@ -3,7 +3,6 @@
 import com.jashmore.sqs.processor.MessageProcessor;
 import com.jashmore.sqs.retriever.MessageRetriever;
 
-import java.util.concurrent.Future;
 import javax.annotation.concurrent.ThreadSafe;
 
 /**
@@ -16,59 +15,6 @@
  * threads. Therefore, implementations of this class must be thread safe.
  */
 @ThreadSafe
-public interface MessageBroker {
-    /**
-     * Start the brokerage of messages from the queue to the processor 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 broker has already been started when this method is called again a {@link IllegalStateException} will be thrown until a subsequent
-     *         {@link #stop()} or {@link #stopWithChildrenThreadsInterrupted()} has been called.</li>
-     *     <li>If this broker is being stopped by calling {@link #stop()} or {@link #stopWithChildrenThreadsInterrupted()}, a call to this
-     *         method should not be blocked by the previous thread and should start a new thread.</li>
-     * </ul>
-     *
-     * @throws IllegalStateException if the broker has already been started
-     */
-    void start();
+public interface MessageBroker extends Runnable {
 
-    /**
-     * Stop the brokerage of messages from the queue to the processor, returning the future that will be resolved when that shutdown is complete.
-     *
-     * <p>Requirements for this method include:
-     *
-     * <ul>
-     *     <li>The broker background thread and any threads created by this background thread must be completed before the {@link Future} returned
-     *         is resolved.</li>
-     *     <li>The children threads processing the messages will not be interrupted and therefore will fully process the message before the {@link Future}
-     *         is resolved.</li>
-     *     <li>If this broker has not been started or has already been stopped, any calls to this method will throw an {@link IllegalStateException}.</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 a future that will resolve when the message broker background thread and all child threads have been stopped/completed
-     * @throws IllegalStateException if the broker has not been started or has already stopped
-     */
-    Future<Object> stop();
-
-    /**
-     * Stop the brokerage of messages from the queue to the processor, returning the future that will be resolved when that shutdown is complete.
-     *
-     * <p>Requirements for this method include:
-     *
-     * <ul>
-     *     <li>The broker background thread and any threads created by this background thread must be completed before the {@link Future} returned
-     *         is resolved.</li>
-     *     <li>The children threads processing the messages will be interrupted and therefore may not fully complete processing the messages before the
-     *         {@link Future} is resolved.</li>
-     *     <li>If this broker has not been started or has already been stopped, any calls to this method will throw an {@link IllegalStateException}.</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 a future that will resolve when the message broker and any child threads have been stopped
-     * @throws IllegalStateException if the broker has not been started or has already stopped
-     */
-    Future<Object> stopWithChildrenThreadsInterrupted();
 }

java-dynamic-sqs-listener-api/src/main/java/com/jashmore/sqs/container/MessageListenerContainer.java

@@ -9,8 +9,7 @@
  * Container used to handle the entire lifecycle of a wrapped method in the spring context.
  *
  * <p>The responsibility of this container is being able to start and stop the processing of messages for the wrapped method. This
- * may only need to call down to the underlying {@link MessageBroker#start()} but there could be more complicated actions needing
- * to be made, such as starting an {@link AsyncMessageRetriever}.
+ * may will start any background threads needed for any component of the library.
  *
  * <p>These containers must be thread safe as there could be multiple threads starting and stopping these containers.
  */

java-dynamic-sqs-listener-api/src/main/java/com/jashmore/sqs/resolver/MessageResolvedResponse.java

@@ -1,4 +1,4 @@
-package com.jashmore.sqs.broker.concurrent.properties;
+package com.jashmore.sqs.broker.concurrent;
 
 import com.google.common.cache.CacheBuilder;
 import com.google.common.cache.CacheLoader;