Merge pull request #86 from JaidenAshmore/issue/85_queue_listener_methods_can_return_completable_futures

JaidenAshmore · web-flow · commit bdb54f68c7ba · 2019-05-14T22:35:21.000-05:00
refs #85: added the ability for queue listener methods to return a completable future
diff --git a/java-dynamic-sqs-listener-api/src/main/java/com/jashmore/sqs/processor/MessageProcessor.java b/java-dynamic-sqs-listener-api/src/main/java/com/jashmore/sqs/processor/MessageProcessor.java
@@ -3,16 +3,28 @@
 import com.jashmore.sqs.argument.ArgumentResolverService;
 import com.jashmore.sqs.broker.MessageBroker;
 import com.jashmore.sqs.processor.argument.Acknowledge;
+import com.jashmore.sqs.resolver.MessageResolver;
 import software.amazon.awssdk.services.sqs.model.Message;
 
+import java.util.concurrent.CompletableFuture;
 import javax.annotation.concurrent.ThreadSafe;
 
 /**
  * Processor that has the responsibility of taking a message and processing it via the required message consumer (Java method).
  *
- * <p>This would therefore need to know what message consumer (method) should be executed for this message and how to execute it.
- * During the message processing it may need to extract data out of the message to populate the parameters of the method with arguments
- * fulfilling what is required. For example, an argument may require a parameter to contain the message id of the message
+ * <p>This would therefore need to know what message consumer (method) should be executed for this message and how to execute it. The default
+ * behaviour of the processor should be if the method processing the method completes without an exception, the message should be resolved by
+ * a {@link MessageResolver}.  However, if the method throws an exception the message processing will be considered a failure and will not be resolved.
+ *
+ * <p>Instead of the approach above of using an exception to consider failures you can have the Java method return a {@link CompletableFuture} and
+ * if that is resolved the message should be considered completed successfully and should be resolved. However, if the {@link CompletableFuture}
+ * is rejected the message should not be resolved, just the same as if an exception was thrown.
+ *
+ * <p>However, if an {@link Acknowledge} parameter is included in the method signature, neither scenarios above should resolve the message and only
+ * calls to {@link Acknowledge#acknowledgeSuccessful()} should resolve the message.
+ *
+ * <p>The parameters of the function will need to extract data out of the message and the implementations of this {@link MessageProcessor} should use
+ * the {@link ArgumentResolverService}. For example, an argument may require a parameter to contain the message id of the message
  * and therefore this would handle populating the argument for this parameter with this value.
  *
  * <p>Most arguments would be able to be resolved via the {@link ArgumentResolverService}, however, the following arguments must be resolved via
diff --git a/java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/processor/DefaultMessageProcessor.java b/java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/processor/DefaultMessageProcessor.java
@@ -12,7 +12,9 @@
 
 import java.lang.reflect.Method;
 import java.lang.reflect.Parameter;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.Arrays;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 import java.util.stream.IntStream;
 import javax.annotation.concurrent.ThreadSafe;
 
@@ -31,16 +33,67 @@ public class DefaultMessageProcessor implements MessageProcessor {
 
     @Override
     public void processMessage(final Message message) throws MessageProcessingException {
-        final Parameter[] parameters = messageConsumerMethod.getParameters();
-
-        final AtomicBoolean hasAcknowledgeField = new AtomicBoolean();
         final Acknowledge acknowledge = () -> messageResolver.resolveMessage(message);
-        final Object[] arguments = IntStream.range(0, parameters.length)
+        final Object[] arguments = getArguments(acknowledge, message);
+
+        final Object result;
+        try {
+            result = messageConsumerMethod.invoke(messageConsumerBean, arguments);
+        } catch (final Throwable throwable) {
+            throw new MessageProcessingException("Error processing message", throwable);
+        }
+
+        if (hasAcknowledgeParameter()) {
+            // If the method has the Acknowledge parameter it is up to them to resolve the message
+            return;
+        }
+
+        final Class<?> returnType = messageConsumerMethod.getReturnType();
+        if (CompletableFuture.class.isAssignableFrom(returnType)) {
+            final CompletableFuture<?> resultCompletableFuture = (CompletableFuture) result;
+
+            if (resultCompletableFuture == null) {
+                throw new MessageProcessingException("Method returns CompletableFuture but null was returned");
+            }
+
+            try {
+                resultCompletableFuture
+                        .thenAccept((ignored) -> acknowledge.acknowledgeSuccessful())
+                        .get();
+            } catch (final InterruptedException interruptedException) {
+                Thread.currentThread().interrupt();
+                throw new MessageProcessingException("Thread interrupted while processing message");
+            } catch (final ExecutionException executionException) {
+                throw new MessageProcessingException("Error processing message", executionException.getCause());
+            }
+        } else {
+            acknowledge.acknowledgeSuccessful();
+        }
+    }
+
+    private boolean hasAcknowledgeParameter() {
+        return Arrays.stream(messageConsumerMethod.getParameters())
+                .anyMatch(DefaultMessageProcessor::isAcknowledgeParameter);
+    }
+
+    private static boolean isAcknowledgeParameter(final Parameter parameter) {
+        return Acknowledge.class.isAssignableFrom(parameter.getType());
+    }
+
+    /**
+     * Get the arguments for the method for the message that is being processed.
+     *
+     * @param acknowledge the acknowledge object that should be used if a parameter is an {@link Acknowledge}
+     * @param message     the message to populate the arguments from
+     * @return the array of arguments to call the method with
+     */
+    private Object[] getArguments(final Acknowledge acknowledge, final Message message) {
+        final Parameter[] parameters = messageConsumerMethod.getParameters();
+        return IntStream.range(0, parameters.length)
                 .mapToObj(parameterIndex -> {
-                    final Parameter parameter = messageConsumerMethod.getParameters()[parameterIndex];
+                    final Parameter parameter = parameters[parameterIndex];
 
-                    if (Acknowledge.class.isAssignableFrom(parameter.getType())) {
-                        hasAcknowledgeField.set(true);
+                    if (isAcknowledgeParameter(parameter)) {
                         return acknowledge;
                     }
 
@@ -58,15 +111,5 @@ public void processMessage(final Message message) throws MessageProcessingExcept
                 })
                 .toArray(Object[]::new);
 
-        try {
-            messageConsumerMethod.invoke(messageConsumerBean, arguments);
-        } catch (final Throwable throwable) {
-            throw new MessageProcessingException("Error processing message", throwable);
-        }
-
-        // If the method doesn't consume the Acknowledge field, it will acknowledge the method here on success
-        if (!hasAcknowledgeField.get()) {
-            acknowledge.acknowledgeSuccessful();
-        }
     }
 }
diff --git a/java-dynamic-sqs-listener-core/src/test/java/com/jashmore/sqs/processor/DefaultMessageProcessorTest.java b/java-dynamic-sqs-listener-core/src/test/java/com/jashmore/sqs/processor/DefaultMessageProcessorTest.java
@@ -6,6 +6,7 @@
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.timeout;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -19,21 +20,32 @@
 import com.jashmore.sqs.resolver.MessageResolver;
 import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.ExpectedException;
 import org.mockito.ArgumentMatchers;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnit;
 import org.mockito.junit.MockitoRule;
 import software.amazon.awssdk.services.sqs.model.Message;
 
 import java.lang.reflect.Method;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
+@SuppressWarnings( {"WeakerAccess", "unused"})
 public class DefaultMessageProcessorTest {
-    private static final String QUEUE_URL = "queueUrl";
+    private static final QueueProperties QUEUE_PROPERTIES = QueueProperties
+            .builder()
+            .queueUrl("queueUrl")
+            .build();
     private static final DefaultMessageProcessorTest BEAN = new DefaultMessageProcessorTest();
 
     @Rule
     public MockitoRule mockitoRule = MockitoJUnit.rule();
 
+    @Rule
+    public ExpectedException expectedException = ExpectedException.none();
+
     @Mock
     private ArgumentResolverService argumentResolverService;
 
@@ -44,34 +56,26 @@ public class DefaultMessageProcessorTest {
     public void forEachParameterInMethodTheArgumentIsResolved() {
         // arrange
         final Method method = getMethodWithAcknowledge();
-        final QueueProperties queueProperties = QueueProperties
-                .builder()
-                .queueUrl(QUEUE_URL)
-                .build();
         final Message message = Message.builder().build();
-        final MessageProcessor processor = new DefaultMessageProcessor(argumentResolverService, queueProperties, messageResolver, method, BEAN);
-        when(argumentResolverService.resolveArgument(eq(queueProperties), any(MethodParameter.class), eq(message)))
+        final MessageProcessor processor = new DefaultMessageProcessor(argumentResolverService, QUEUE_PROPERTIES, messageResolver, method, BEAN);
+        when(argumentResolverService.resolveArgument(eq(QUEUE_PROPERTIES), any(MethodParameter.class), eq(message)))
                 .thenReturn("payload")
                 .thenReturn("payload2");
 
         // act
         processor.processMessage(message);
 
         // assert
-        verify(argumentResolverService, times(2)).resolveArgument(eq(queueProperties), any(MethodParameter.class), eq(message));
+        verify(argumentResolverService, times(2)).resolveArgument(eq(QUEUE_PROPERTIES), any(MethodParameter.class), eq(message));
     }
 
     @Test
     public void anyParameterUnableToBeResolvedWillThrowAnError() {
         // arrange
         final Method method = getMethodWithAcknowledge();
-        final QueueProperties queueProperties = QueueProperties
-                .builder()
-                .queueUrl(QUEUE_URL)
-                .build();
         final Message message = Message.builder().build();
-        final MessageProcessor processor = new DefaultMessageProcessor(argumentResolverService, queueProperties, messageResolver, method, BEAN);
-        when(argumentResolverService.resolveArgument(eq(queueProperties), any(MethodParameter.class), eq(message)))
+        final MessageProcessor processor = new DefaultMessageProcessor(argumentResolverService, QUEUE_PROPERTIES, messageResolver, method, BEAN);
+        when(argumentResolverService.resolveArgument(eq(QUEUE_PROPERTIES), any(MethodParameter.class), eq(message)))
                 .thenThrow(new ArgumentResolutionException("Error resolving"));
 
         // act
@@ -89,13 +93,9 @@ public void methodWillBeInvokedWithArgumentsResolved() {
         // arrange
         final Method method = getMethodWithAcknowledge();
         final DefaultMessageProcessorTest mockProcessor = mock(DefaultMessageProcessorTest.class);
-        final QueueProperties queueProperties = QueueProperties
-                .builder()
-                .queueUrl(QUEUE_URL)
-                .build();
         final Message message = Message.builder().build();
-        final MessageProcessor processor = new DefaultMessageProcessor(argumentResolverService, queueProperties, messageResolver, method, mockProcessor);
-        when(argumentResolverService.resolveArgument(eq(queueProperties), any(MethodParameter.class), eq(message)))
+        final MessageProcessor processor = new DefaultMessageProcessor(argumentResolverService, QUEUE_PROPERTIES, messageResolver, method, mockProcessor);
+        when(argumentResolverService.resolveArgument(eq(QUEUE_PROPERTIES), any(MethodParameter.class), eq(message)))
                 .thenReturn("payload")
                 .thenReturn("payload2");
 
@@ -110,13 +110,9 @@ public void methodWillBeInvokedWithArgumentsResolved() {
     public void methodWithAcknowledgeParameterWillNotDeleteMessageOnSuccess() {
         // arrange
         final Method method = getMethodWithAcknowledge();
-        final QueueProperties queueProperties = QueueProperties
-                .builder()
-                .queueUrl(QUEUE_URL)
-                .build();
         final Message message = Message.builder().build();
-        final MessageProcessor processor = new DefaultMessageProcessor(argumentResolverService, queueProperties, messageResolver, method, BEAN);
-        when(argumentResolverService.resolveArgument(eq(queueProperties), any(MethodParameter.class), eq(message)))
+        final MessageProcessor processor = new DefaultMessageProcessor(argumentResolverService, QUEUE_PROPERTIES, messageResolver, method, BEAN);
+        when(argumentResolverService.resolveArgument(eq(QUEUE_PROPERTIES), any(MethodParameter.class), eq(message)))
                 .thenReturn("payload");
 
         // act
@@ -130,13 +126,9 @@ public void methodWithAcknowledgeParameterWillNotDeleteMessageOnSuccess() {
     public void methodWithoutAcknowledgeParameterWillDeleteMessageOnSuccess() {
         // arrange
         final Method method = getMethodWithNoAcknowledge();
-        final QueueProperties queueProperties = QueueProperties
-                .builder()
-                .queueUrl(QUEUE_URL)
-                .build();
         final Message message = Message.builder().receiptHandle("handle").build();
-        final MessageProcessor processor = new DefaultMessageProcessor(argumentResolverService, queueProperties, messageResolver, method, BEAN);
-        when(argumentResolverService.resolveArgument(eq(queueProperties), any(MethodParameter.class), eq(message)))
+        final MessageProcessor processor = new DefaultMessageProcessor(argumentResolverService, QUEUE_PROPERTIES, messageResolver, method, BEAN);
+        when(argumentResolverService.resolveArgument(eq(QUEUE_PROPERTIES), any(MethodParameter.class), eq(message)))
                 .thenReturn("payload");
 
         // act
@@ -151,13 +143,9 @@ public void methodWithoutAcknowledgeParameterWillDeleteMessageOnSuccess() {
     public void methodWithoutAcknowledgeThatThrowsExceptionDoesNotDeleteMessage() {
         // arrange
         final Method method = getMethodThatThrowsException();
-        final QueueProperties queueProperties = QueueProperties
-                .builder()
-                .queueUrl(QUEUE_URL)
-                .build();
         final Message message = Message.builder().receiptHandle("handle").build();
-        final MessageProcessor processor = new DefaultMessageProcessor(argumentResolverService, queueProperties, messageResolver, method, BEAN);
-        when(argumentResolverService.resolveArgument(eq(queueProperties), any(MethodParameter.class), eq(message)))
+        final MessageProcessor processor = new DefaultMessageProcessor(argumentResolverService, QUEUE_PROPERTIES, messageResolver, method, BEAN);
+        when(argumentResolverService.resolveArgument(eq(QUEUE_PROPERTIES), any(MethodParameter.class), eq(message)))
                 .thenReturn("payload");
 
         // act
@@ -170,17 +158,95 @@ public void methodWithoutAcknowledgeThatThrowsExceptionDoesNotDeleteMessage() {
         }
     }
 
-    @SuppressWarnings("unused")
+    @Test
+    public void methodReturningCompletableFutureWillResolveMessageWhenFutureResolved() throws Exception {
+        // arrange
+        final Method method = DefaultMessageProcessorTest.class.getMethod("methodReturningCompletableFuture", CompletableFuture.class);
+        final MessageProcessor processor = new DefaultMessageProcessor(argumentResolverService, QUEUE_PROPERTIES, messageResolver, method, BEAN);
+        final Message message = Message.builder().receiptHandle("handle").build();
+        final CompletableFuture<Object> future = new CompletableFuture<>();
+        when(argumentResolverService.resolveArgument(eq(QUEUE_PROPERTIES), any(MethodParameter.class), eq(message)))
+                .thenReturn(future);
+
+        // act
+        CompletableFuture.runAsync(() -> processor.processMessage(message));
+
+        // assert
+        verify(messageResolver, never()).resolveMessage(message);
+        future.complete("value");
+        verify(messageResolver, timeout(100)).resolveMessage(message);
+    }
+
+    @Test
+    public void methodReturningCompletableFutureWillNotResolveMessageWhenFutureRejected() throws Exception {
+        // arrange
+        final Method method = DefaultMessageProcessorTest.class.getMethod("methodReturningCompletableFuture", CompletableFuture.class);
+        final MessageProcessor processor = new DefaultMessageProcessor(argumentResolverService, QUEUE_PROPERTIES, messageResolver, method, BEAN);
+        final Message message = Message.builder().receiptHandle("handle").build();
+        final CompletableFuture<Object> future = new CompletableFuture<>();
+        when(argumentResolverService.resolveArgument(eq(QUEUE_PROPERTIES), any(MethodParameter.class), eq(message)))
+                .thenReturn(future);
+
+        // act
+        CompletableFuture.runAsync(() -> processor.processMessage(message));
+        future.completeExceptionally(new RuntimeException("Excepted Exception"));
+        Thread.sleep(100); // make sure it doesn't resolve
+
+        // assert
+        verify(messageResolver, never()).resolveMessage(message);
+    }
+
+    @Test
+    public void methodReturningCompletableFutureThatReturnsNullWillThrowMessageProcessingException() throws Exception {
+        // arrange
+        final Method method = DefaultMessageProcessorTest.class.getMethod("methodReturningCompletableFuture", CompletableFuture.class);
+        final MessageProcessor processor = new DefaultMessageProcessor(argumentResolverService, QUEUE_PROPERTIES, messageResolver, method, BEAN);
+        final Message message = Message.builder().receiptHandle("handle").build();
+        when(argumentResolverService.resolveArgument(eq(QUEUE_PROPERTIES), any(MethodParameter.class), eq(message)))
+                .thenReturn(null);
+        expectedException.expect(MessageProcessingException.class);
+
+        // act
+        processor.processMessage(message);
+    }
+
+    @Test
+    public void threadInterruptedWhileGettingMessageWillThrowException() throws Exception {
+        // arrange
+        final Method method = DefaultMessageProcessorTest.class.getMethod("methodReturningCompletableFuture", CompletableFuture.class);
+        final MessageProcessor processor = new DefaultMessageProcessor(argumentResolverService, QUEUE_PROPERTIES, messageResolver, method, BEAN);
+        final Message message = Message.builder().receiptHandle("handle").build();
+        final CompletableFuture<Object> future = new CompletableFuture<>();
+        when(argumentResolverService.resolveArgument(eq(QUEUE_PROPERTIES), any(MethodParameter.class), eq(message)))
+                .thenReturn(future);
+
+        // act
+        final AtomicBoolean exceptionThrown = new AtomicBoolean(false);
+        CompletableFuture.runAsync(() -> {
+            try {
+                Thread.currentThread().interrupt(); // interrupt the thread so the InterruptedException will be thrown while blocking on the future
+                processor.processMessage(message);
+            } catch (final MessageProcessingException messageProcessingException) {
+                exceptionThrown.set(true);
+            }
+        }).get(1, TimeUnit.SECONDS);
+
+        // assert
+        assertThat(exceptionThrown).isTrue();
+    }
+
     public void methodWithNoAcknowledge(@Payload String payload, @Payload String payloadTwo) {
 
     }
 
-    @SuppressWarnings( {"unused", "WeakerAccess"})
     public void methodWithAcknowledge(Acknowledge acknowledge, @Payload String payload, @Payload String payloadTwo) {
 
     }
 
-    @SuppressWarnings("unused")
+    public CompletableFuture<?> methodReturningCompletableFuture(CompletableFuture<?> futureToReturn) {
+        return futureToReturn;
+    }
+
     public void methodThatThrowsException(@Payload String payload) {
         throw new RuntimeException("error");
     }
diff --git a/pom.xml b/pom.xml
