AsyncSqsClientImpl implements AsyncSqsClient { @Override public CompletableFuture<Boolean> consume( @NonNull final Consumer<String> consumer, @NonNull final String queueName) { return consumeInternal( (body, executor) -> CompletableFuture.runAsync(() -> consumer.accept(body), executor), queueName); } @Inject  AsyncSqsClientImpl(
      @NonNull final SqsAsyncClient client,
      @NonNull final AsyncConveyorMetrics metrics,
      @NonNull final ExecutorService executor);   AsyncSqsClientImpl(
      @NonNull final SqsAsyncClient client,
      @NonNull final AsyncConveyorMetrics metrics,
      @NonNull final ExecutorService executor,
      long maxCacheSize,
      int receiveWaitTimeoutSeconds,
      int bulkheadMaxWaitMillis,
      int consumerConcurrency); @Override CompletableFuture<Void> add(
      @NonNull final String message, @NonNull final String queueName); @Override CompletableFuture<Void> add(
      @NonNull final String message, @NonNull final String queueName, int delaySeconds); @Override CompletableFuture<Boolean> consume(
      @NonNull final Consumer<String> consumer, @NonNull final String queueName); @Override CompletableFuture<Boolean> consume(AsyncConsumer<String> consumer, String queueName); @Override void close();  }
@Test(expected = Exception.class) public void testConsumeInternalsFailure() { urlSetup("testQueue", "testUrl"); receiveSetup(); ChangeMessageVisibilityRequest changeVisibilityReq = ChangeMessageVisibilityRequest.builder() .queueUrl("testUrl") .receiptHandle("testHandle") .visibilityTimeout(0) .build(); CompletableFuture<ChangeMessageVisibilityResponse> visibilityResponse = CompletableFuture.completedFuture(ChangeMessageVisibilityResponse.builder().build()); when(awsAsyncSqsClient.changeMessageVisibility(changeVisibilityReq)) .thenReturn(visibilityResponse); ArgumentCaptor<ChangeMessageVisibilityRequest> visibilityReq = ArgumentCaptor.forClass(ChangeMessageVisibilityRequest.class); CompletableFuture<Void> future = new CompletableFuture<>(); future.completeExceptionally(new Exception("async consumer completing exceptionally")); AsyncConsumer<String> function = (str, threadpool) -> future; CompletableFuture<Boolean> result = asyncClient.consume(function, "testQueue"); verify(awsAsyncSqsClient).changeMessageVisibility(visibilityReq.capture()); assertEquals(visibilityReq.getValue().queueUrl(), "testUrl"); assertEquals(visibilityReq.getValue().receiptHandle(), "testHandle"); assertEquals(visibilityReq.getValue().visibilityTimeout(), new Integer(0)); verify(awsAsyncSqsClient, times(1)) .changeMessageVisibility(any(ChangeMessageVisibilityRequest.class)); result.join(); }
@Test public void testConsumeEmptyQueue() { urlSetup("empty", "emptyUrl"); CompletableFuture<ReceiveMessageResponse> noMessages = CompletableFuture.completedFuture(ReceiveMessageResponse.builder().build()); ReceiveMessageRequest testEmptyRequest = ReceiveMessageRequest.builder() .queueUrl("emptyUrl") .waitTimeSeconds(AsyncSqsClientConfiguration.DEFAULT_RECEIVE_WAIT_SECONDS) .maxNumberOfMessages(1) .build(); when(awsAsyncSqsClient.receiveMessage(testEmptyRequest)).thenReturn(noMessages); Consumer<String> function = str -> System.out.println(str); CompletableFuture<Boolean> response = asyncClient.consume(function, "empty"); verify(awsAsyncSqsClient, times(1)).receiveMessage(any(ReceiveMessageRequest.class)); assertFalse(response.join()); }
@Test public void testAsyncConsumeConsumption() { urlSetup("testQueue", "testUrl"); receiveSetup(); deleteSetup(); AsyncConsumer<String> function = (str, threadpool) -> CompletableFuture.completedFuture(null); CompletableFuture<Boolean> result = asyncClient.consume(function, "testQueue"); assertTrue(result.join()); }
@Test public void testConsumeCompletionWithReceiveFailure() { urlSetup("receiveFailure", "receiveFailureUrl"); CompletableFuture<ReceiveMessageResponse> receiveMessage = new CompletableFuture<>(); receiveMessage.completeExceptionally(new Exception()); ReceiveMessageRequest testRequest = ReceiveMessageRequest.builder() .queueUrl("receiveFailureUrl") .maxNumberOfMessages(1) .waitTimeSeconds(AsyncSqsClientConfiguration.DEFAULT_RECEIVE_WAIT_SECONDS) .build(); when(awsAsyncSqsClient.receiveMessage(testRequest)).thenReturn(receiveMessage); Consumer<String> function = System.out::println; CompletableFuture<Boolean> result = asyncClient.consume(function, "receiveFailure"); try { result.get(1000, TimeUnit.MICROSECONDS); } catch (TimeoutException timeout) { fail("Future does not seem to complete when failure in receiveMessage."); } catch (Exception ex) { } urlFailureSetup("receiveFailure"); result = asyncClient.consume(function, "receiveFailure"); try { result.get(1000, TimeUnit.MILLISECONDS); } catch (TimeoutException timeout) { fail("Future does not seem to complete when failure in getQueueUrl within receiveMessage."); } catch (Exception ex) { } }
@Test public void testConsumeCompletionWithChangeVisibilityFailure() { urlSetup("visibilityFailure", "visibilityFailureUrl"); ChangeMessageVisibilityRequest changeVisibilityReq = ChangeMessageVisibilityRequest.builder() .queueUrl("visibilityFailureUrl") .receiptHandle("visibilityFailure") .visibilityTimeout(0) .build(); CompletableFuture<ChangeMessageVisibilityResponse> visibilityResponse = new CompletableFuture<>(); visibilityResponse.completeExceptionally(new Exception()); when(awsAsyncSqsClient.changeMessageVisibility(changeVisibilityReq)) .thenReturn(visibilityResponse); CompletableFuture<Void> future = new CompletableFuture<>(); future.completeExceptionally(new Exception("async consumer completing exceptionally")); AsyncConsumer<String> function = (str, threadpool) -> future; CompletableFuture<Boolean> result = asyncClient.consume(function, "visibilityFailure"); try { result.get(1000, TimeUnit.MICROSECONDS); } catch (TimeoutException timeout) { fail("Future does not seem to complete when failure in changeMessageVisibility."); } catch (Exception ex) { } urlFailureSetup("visibilityFailure"); result = asyncClient.consume(function, "visibilityFailure"); try { result.get(1000, TimeUnit.MILLISECONDS); } catch (TimeoutException timeout) { fail( "Future does not seem to complete when failure in getQueueUrl within changeMessageVisibility."); } catch (Exception ex) { } }