SQSPoller { public void poll(final int remainingTimeInMillis) { Instant cutoff = Instant.now(clock) .plusMillis(remainingTimeInMillis) .minusMillis(TIMEOUT_BUFFER_IN_MILLIS); messageDispatcher.reset(); int estimatedCapacity; while ((estimatedCapacity = messageDispatcher.getEstimatedCapacity(cutoff)) > 0) { List<Message> toProcess = sqsProxy.receiveMessages(estimatedCapacity); if (toProcess.isEmpty()) { log.info("No messages received from queue. Returning until next polling cycle to save cost."); return; } messageDispatcher.dispatch(toProcess); } } SQSPoller(final SQSProxy sqsProxy, final MessageDispatcher messageDispatcher); void poll(final int remainingTimeInMillis);  }
@Test public void poll_messagesReceived() throws Exception { when(messageDispatcher.getEstimatedCapacity(any())) .thenReturn(4) .thenReturn(2) .thenReturn(0); List<Message> messages = Lists.newArrayList(mock(Message.class), mock(Message.class)); when(sqsProxy.receiveMessages(anyInt())).thenReturn(messages); poller.poll(SQSPoller.TIMEOUT_BUFFER_IN_MILLIS * 2); Instant expectedCutoff = NOW.plusMillis(SQSPoller.TIMEOUT_BUFFER_IN_MILLIS); InOrder inOrder = inOrder(messageDispatcher, sqsProxy); inOrder.verify(messageDispatcher).reset(); inOrder.verify(messageDispatcher).getEstimatedCapacity(expectedCutoff); inOrder.verify(sqsProxy).receiveMessages(4); inOrder.verify(messageDispatcher).dispatch(messages); inOrder.verify(messageDispatcher).getEstimatedCapacity(expectedCutoff); inOrder.verify(sqsProxy).receiveMessages(2); inOrder.verify(messageDispatcher).dispatch(messages); inOrder.verify(messageDispatcher).getEstimatedCapacity(expectedCutoff); verifyNoMoreInteractions(sqsProxy, messageDispatcher); }
@Test public void poll_noMessagesReceived() throws Exception { when(messageDispatcher.getEstimatedCapacity(any())).thenReturn(1); when(sqsProxy.receiveMessages(anyInt())).thenReturn(Collections.emptyList()); poller.poll(SQSPoller.TIMEOUT_BUFFER_IN_MILLIS * 2); verify(messageDispatcher, never()).dispatch(any()); }