SQSMessageProcessor implements Function<SQSMessageProcessorRequest, SQSMessageProcessorResponse> { @Override public SQSMessageProcessorResponse apply(final SQSMessageProcessorRequest request) { List<SQSMessageResult> messageResults = request.getMessages().stream() .map(this::processMessage) .filter(Objects::nonNull) .collect(Collectors.toList()); return new SQSMessageProcessorResponse(messageResults); }  @Override SQSMessageProcessorResponse apply(final SQSMessageProcessorRequest request);  }
@Test public void apply() throws Exception { doNothing() .doThrow(new RetryMessageException("try again later...").withRetryDelayInSeconds(60)) .doThrow(RuntimeException.class) .when(delegate).accept(any(Foo.class)); Message m1 = new Message().withMessageId("1"); Message m2 = new Message().withMessageId("2"); Message m3 = new Message().withMessageId("3"); SQSMessageProcessorRequest request = new SQSMessageProcessorRequest(Lists.newArrayList(m1, m2, m3)); SQSMessageProcessorResponse response = messageProcessor.apply(request); SQSMessageProcessorResponse expected = new SQSMessageProcessorResponse(Lists.newArrayList( new SQSMessageResult("1", SQSMessageResult.Status.SUCCESS), new SQSMessageResult("2", SQSMessageResult.Status.RETRY, 60), new SQSMessageResult("3", SQSMessageResult.Status.ERROR) )); assertThat(response, is(expected)); }