SqsMessagePoller { @Scheduled(fixedDelay = 1) public void loop() { List<Message> messages = reveiveMessages(); if (messages.isEmpty()) { log.trace("No SQS message received"); return; } log.debug("{} SQS messages are received", messages.size()); messages.stream().parallel().forEach(this::handleMessage); }  @Scheduled(fixedDelay = 1) // SUPPRESS CHECKSTYLE bug? void loop();  }
@Test public void test_0Messages() throws Exception { when(sqs.receiveMessage(any(ReceiveMessageRequest.class))) .thenReturn(receiveMessageResultOf()); sut.loop(); verify(sqs).receiveMessage(any(ReceiveMessageRequest.class)); verify(messageHandler, never()).accept(any(Message.class)); verify(sqs, never()).deleteMessage(any(DeleteMessageRequest.class)); verify(sqs, never()).changeMessageVisibility(any(ChangeMessageVisibilityRequest.class)); }
@Test public void test_1Message() throws Exception { Message msg1 = createMessage(1); DeleteMessageRequest expectedDmr = createDeleteMessageRequest(1); when(sqs.receiveMessage(any(ReceiveMessageRequest.class))).thenReturn(receiveMessageResultOf(msg1)); sut.loop(); verify(sqs).receiveMessage(any(ReceiveMessageRequest.class)); verify(messageHandler).accept(eq(msg1)); verify(sqs).deleteMessage(eq(expectedDmr)); verify(sqs, never()).changeMessageVisibility(any(ChangeMessageVisibilityRequest.class)); }
@Test public void test_3Message() throws Exception { Message msg1 = createMessage(1); Message msg2 = createMessage(2); Message msg3 = createMessage(3); DeleteMessageRequest expectedDmr1 = createDeleteMessageRequest(1); DeleteMessageRequest expectedDmr2 = createDeleteMessageRequest(2); DeleteMessageRequest expectedDmr3 = createDeleteMessageRequest(3); when(sqs.receiveMessage(any(ReceiveMessageRequest.class))) .thenReturn(receiveMessageResultOf(msg1, msg2, msg3)); sut.loop(); verify(sqs).receiveMessage(any(ReceiveMessageRequest.class)); verify(messageHandler).accept(eq(msg1)); verify(messageHandler).accept(eq(msg2)); verify(messageHandler).accept(eq(msg3)); verify(sqs).deleteMessage(eq(expectedDmr1)); verify(sqs).deleteMessage(eq(expectedDmr2)); verify(sqs).deleteMessage(eq(expectedDmr3)); verify(sqs, never()).changeMessageVisibility(any(ChangeMessageVisibilityRequest.class)); }
@Test public void test_1HeavyMessage() throws Exception { Message msg1 = createMessage(1); DeleteMessageRequest expectedDmr = createDeleteMessageRequest(1); when(sqs.receiveMessage(any(ReceiveMessageRequest.class))).thenReturn(receiveMessageResultOf(msg1)); doAnswer(createHeavyJobAnswer(4, false)).when(messageHandler).accept(any(Message.class)); sut.loop(); verify(sqs).receiveMessage(any(ReceiveMessageRequest.class)); verify(messageHandler).accept(eq(msg1)); verify(sqs).deleteMessage(eq(expectedDmr)); ArgumentCaptor<ChangeMessageVisibilityRequest> captor = ArgumentCaptor.forClass(ChangeMessageVisibilityRequest.class); verify(sqs, times(1)).changeMessageVisibility(captor.capture()); ChangeMessageVisibilityRequest cmvReq = captor.getValue(); assertThat(cmvReq.getQueueUrl(), is(Q_URL)); assertThat(cmvReq.getReceiptHandle(), is("rh-1")); assertThat(cmvReq.getVisibilityTimeout(), is(10)); }
@Test public void test_3HeavyMessage() throws Exception { Message msg1 = createMessage(1); Message msg2 = createMessage(2); Message msg3 = createMessage(3); DeleteMessageRequest expectedDmr1 = createDeleteMessageRequest(1); DeleteMessageRequest expectedDmr2 = createDeleteMessageRequest(2); DeleteMessageRequest expectedDmr3 = createDeleteMessageRequest(3); doReturn(receiveMessageResultOf(msg1, msg2, msg3)) .when(sqs).receiveMessage(any(ReceiveMessageRequest.class)); doAnswer(createHeavyJobAnswer(4, false)) .when(messageHandler).accept(any(Message.class)); sut.loop(); verify(sqs).receiveMessage(any(ReceiveMessageRequest.class)); verify(messageHandler).accept(eq(msg1)); verify(messageHandler).accept(eq(msg2)); verify(messageHandler).accept(eq(msg3)); verify(sqs).deleteMessage(eq(expectedDmr1)); verify(sqs).deleteMessage(eq(expectedDmr2)); verify(sqs).deleteMessage(eq(expectedDmr3)); ArgumentCaptor<ChangeMessageVisibilityRequest> captor = ArgumentCaptor.forClass(ChangeMessageVisibilityRequest.class); verify(sqs, times(3)).changeMessageVisibility(captor.capture()); ChangeMessageVisibilityRequest cmvReq1 = captor.getAllValues().stream() .filter(r -> r.getReceiptHandle().equals("rh-1")).findFirst().get(); assertThat(cmvReq1.getQueueUrl(), is(Q_URL)); assertThat(cmvReq1.getVisibilityTimeout(), is(10)); ChangeMessageVisibilityRequest cmvReq2 = captor.getAllValues().stream() .filter(r -> r.getReceiptHandle().equals("rh-2")).findFirst().get(); assertThat(cmvReq2.getQueueUrl(), is(Q_URL)); assertThat(cmvReq2.getReceiptHandle(), is("rh-2")); assertThat(cmvReq2.getVisibilityTimeout(), is(10)); ChangeMessageVisibilityRequest cmvReq3 = captor.getAllValues().stream() .filter(r -> r.getReceiptHandle().equals("rh-3")).findFirst().get(); assertThat(cmvReq3.getQueueUrl(), is(Q_URL)); assertThat(cmvReq3.getReceiptHandle(), is("rh-3")); assertThat(cmvReq3.getVisibilityTimeout(), is(10)); }
@Test public void test_1VeryHeavyMessage() throws Exception { Message msg1 = createMessage(1); DeleteMessageRequest expectedDmr = createDeleteMessageRequest(1); when(sqs.receiveMessage(any(ReceiveMessageRequest.class))).thenReturn(receiveMessageResultOf(msg1)); doAnswer(createHeavyJobAnswer(7, false)).when(messageHandler).accept(any(Message.class)); sut.loop(); verify(sqs).receiveMessage(any(ReceiveMessageRequest.class)); verify(messageHandler).accept(eq(msg1)); verify(sqs).deleteMessage(eq(expectedDmr)); ArgumentCaptor<ChangeMessageVisibilityRequest> captor = ArgumentCaptor.forClass(ChangeMessageVisibilityRequest.class); verify(sqs, times(2)).changeMessageVisibility(captor.capture()); ChangeMessageVisibilityRequest cmvReq1 = captor.getAllValues().get(0); assertThat(cmvReq1.getQueueUrl(), is(Q_URL)); assertThat(cmvReq1.getReceiptHandle(), is("rh-1")); assertThat(cmvReq1.getVisibilityTimeout(), is(10)); ChangeMessageVisibilityRequest cmvReq2 = captor.getAllValues().get(1); assertThat(cmvReq2.getQueueUrl(), is(Q_URL)); assertThat(cmvReq2.getReceiptHandle(), is("rh-1")); assertThat(cmvReq2.getVisibilityTimeout(), is(10)); }
@Test public void test_1ExcessiveHeavyMessage() throws Exception { sut.setVisibilityTimeout(60); sut.setChangeVisibilityThreshold(1); Message msg1 = createMessage(1); DeleteMessageRequest expectedDmr = createDeleteMessageRequest(1); when(sqs.receiveMessage(any(ReceiveMessageRequest.class))).thenReturn(receiveMessageResultOf(msg1)); doAnswer(createHeavyJobAnswer(24, false)).when(messageHandler).accept(any(Message.class)); sut.loop(); verify(sqs).receiveMessage(any(ReceiveMessageRequest.class)); verify(messageHandler).accept(eq(msg1)); verify(sqs, never()).deleteMessage(eq(expectedDmr)); verify(sqs, times(3)).changeMessageVisibility(any(ChangeMessageVisibilityRequest.class)); }
@Test public void test_1MessageWithFailureHandler() throws Exception { Message msg1 = createMessage(1); DeleteMessageRequest expectedDmr = createDeleteMessageRequest(1); when(sqs.receiveMessage(any(ReceiveMessageRequest.class))).thenReturn(receiveMessageResultOf(msg1)); doThrow(RuntimeException.class).when(messageHandler).accept(any(Message.class)); sut.loop(); verify(sqs).receiveMessage(any(ReceiveMessageRequest.class)); verify(messageHandler).accept(eq(msg1)); verify(sqs, never()).deleteMessage(eq(expectedDmr)); verify(sqs, never()).changeMessageVisibility(any(ChangeMessageVisibilityRequest.class)); }
@Test public void test_1HeavyMessageWithFailureHandler() throws Exception { Message msg1 = createMessage(1); DeleteMessageRequest expectedDmr = createDeleteMessageRequest(1); when(sqs.receiveMessage(any(ReceiveMessageRequest.class))).thenReturn(receiveMessageResultOf(msg1)); doAnswer(createHeavyJobAnswer(4, true)).when(messageHandler).accept(any(Message.class)); sut.loop(); verify(sqs).receiveMessage(any(ReceiveMessageRequest.class)); verify(messageHandler).accept(eq(msg1)); verify(sqs, never()).deleteMessage(eq(expectedDmr)); verify(sqs).changeMessageVisibility(any(ChangeMessageVisibilityRequest.class)); }