@Test public void testNoResults() { ReceiveMessageResult receiveMessageResult = new ReceiveMessageResult(); receiveMessageResult.setMessages(new ArrayList<>()); doReturn(receiveMessageResult).when(mockAmazonSQS).receiveMessage(any(ReceiveMessageRequest.class)); QueueResult result = pollRequestProcessor.processRequest( new SQSRequestContext(mockAmazonSQS, new QueueRequest(), null)); verify(mockAmazonSQS, times(1)).receiveMessage(any(ReceiveMessageRequest.class)); Assert.assertTrue(result.isSuccessful()); Assert.assertNull(result.getData()); }
@Override public SQSQueueResult processRequest(SQSRequestContext sqsRequestContext) { AmazonSQS amazonSQS = sqsRequestContext.getSqsClient(); SQSQueueResult result = new SQSQueueResult(); ReceiveMessageRequest receiveMessageRequest = new ReceiveMessageRequest() .withQueueUrl(sqsRequestContext.getQueueURL()) .withMaxNumberOfMessages(getMaxMessages(sqsRequestContext)); receiveMessageRequest.setWaitTimeSeconds(waitTimeSeconds); ReceiveMessageResult receiveMessageResult = amazonSQS.receiveMessage(receiveMessageRequest); if (receiveMessageResult.getMessages().size() >= 1) { List<String> messageBodies = new LinkedList<>(); for (Message receivedMessage : receiveMessageResult.getMessages()) { try { DeleteMessageRequest deleteMessageRequest = new DeleteMessageRequest() .withQueueUrl(sqsRequestContext.getQueueURL()) .withReceiptHandle(receivedMessage.getReceiptHandle()); amazonSQS.deleteMessage(deleteMessageRequest); messageBodies.add(receivedMessage.getBody()); } catch(Exception e) { } } result .setSuccessful(true) .setData(messageBodies); } else { result.setSuccessful(true); result.setData(null); } return result; }
PollRequestProcessor implements SQSRequestProcessor { @Override public SQSQueueResult processRequest(SQSRequestContext sqsRequestContext) { AmazonSQS amazonSQS = sqsRequestContext.getSqsClient(); SQSQueueResult result = new SQSQueueResult(); ReceiveMessageRequest receiveMessageRequest = new ReceiveMessageRequest() .withQueueUrl(sqsRequestContext.getQueueURL()) .withMaxNumberOfMessages(getMaxMessages(sqsRequestContext)); receiveMessageRequest.setWaitTimeSeconds(waitTimeSeconds); ReceiveMessageResult receiveMessageResult = amazonSQS.receiveMessage(receiveMessageRequest); if (receiveMessageResult.getMessages().size() >= 1) { List<String> messageBodies = new LinkedList<>(); for (Message receivedMessage : receiveMessageResult.getMessages()) { try { DeleteMessageRequest deleteMessageRequest = new DeleteMessageRequest() .withQueueUrl(sqsRequestContext.getQueueURL()) .withReceiptHandle(receivedMessage.getReceiptHandle()); amazonSQS.deleteMessage(deleteMessageRequest); messageBodies.add(receivedMessage.getBody()); } catch(Exception e) { } } result .setSuccessful(true) .setData(messageBodies); } else { result.setSuccessful(true); result.setData(null); } return result; } }
PollRequestProcessor implements SQSRequestProcessor { @Override public SQSQueueResult processRequest(SQSRequestContext sqsRequestContext) { AmazonSQS amazonSQS = sqsRequestContext.getSqsClient(); SQSQueueResult result = new SQSQueueResult(); ReceiveMessageRequest receiveMessageRequest = new ReceiveMessageRequest() .withQueueUrl(sqsRequestContext.getQueueURL()) .withMaxNumberOfMessages(getMaxMessages(sqsRequestContext)); receiveMessageRequest.setWaitTimeSeconds(waitTimeSeconds); ReceiveMessageResult receiveMessageResult = amazonSQS.receiveMessage(receiveMessageRequest); if (receiveMessageResult.getMessages().size() >= 1) { List<String> messageBodies = new LinkedList<>(); for (Message receivedMessage : receiveMessageResult.getMessages()) { try { DeleteMessageRequest deleteMessageRequest = new DeleteMessageRequest() .withQueueUrl(sqsRequestContext.getQueueURL()) .withReceiptHandle(receivedMessage.getReceiptHandle()); amazonSQS.deleteMessage(deleteMessageRequest); messageBodies.add(receivedMessage.getBody()); } catch(Exception e) { } } result .setSuccessful(true) .setData(messageBodies); } else { result.setSuccessful(true); result.setData(null); } return result; }  }
PollRequestProcessor implements SQSRequestProcessor { @Override public SQSQueueResult processRequest(SQSRequestContext sqsRequestContext) { AmazonSQS amazonSQS = sqsRequestContext.getSqsClient(); SQSQueueResult result = new SQSQueueResult(); ReceiveMessageRequest receiveMessageRequest = new ReceiveMessageRequest() .withQueueUrl(sqsRequestContext.getQueueURL()) .withMaxNumberOfMessages(getMaxMessages(sqsRequestContext)); receiveMessageRequest.setWaitTimeSeconds(waitTimeSeconds); ReceiveMessageResult receiveMessageResult = amazonSQS.receiveMessage(receiveMessageRequest); if (receiveMessageResult.getMessages().size() >= 1) { List<String> messageBodies = new LinkedList<>(); for (Message receivedMessage : receiveMessageResult.getMessages()) { try { DeleteMessageRequest deleteMessageRequest = new DeleteMessageRequest() .withQueueUrl(sqsRequestContext.getQueueURL()) .withReceiptHandle(receivedMessage.getReceiptHandle()); amazonSQS.deleteMessage(deleteMessageRequest); messageBodies.add(receivedMessage.getBody()); } catch(Exception e) { } } result .setSuccessful(true) .setData(messageBodies); } else { result.setSuccessful(true); result.setData(null); } return result; }  @Override SQSQueueResult processRequest(SQSRequestContext sqsRequestContext); void setWaitTimeSeconds(int waitTimeSeconds); }
PollRequestProcessor implements SQSRequestProcessor { @Override public SQSQueueResult processRequest(SQSRequestContext sqsRequestContext) { AmazonSQS amazonSQS = sqsRequestContext.getSqsClient(); SQSQueueResult result = new SQSQueueResult(); ReceiveMessageRequest receiveMessageRequest = new ReceiveMessageRequest() .withQueueUrl(sqsRequestContext.getQueueURL()) .withMaxNumberOfMessages(getMaxMessages(sqsRequestContext)); receiveMessageRequest.setWaitTimeSeconds(waitTimeSeconds); ReceiveMessageResult receiveMessageResult = amazonSQS.receiveMessage(receiveMessageRequest); if (receiveMessageResult.getMessages().size() >= 1) { List<String> messageBodies = new LinkedList<>(); for (Message receivedMessage : receiveMessageResult.getMessages()) { try { DeleteMessageRequest deleteMessageRequest = new DeleteMessageRequest() .withQueueUrl(sqsRequestContext.getQueueURL()) .withReceiptHandle(receivedMessage.getReceiptHandle()); amazonSQS.deleteMessage(deleteMessageRequest); messageBodies.add(receivedMessage.getBody()); } catch(Exception e) { } } result .setSuccessful(true) .setData(messageBodies); } else { result.setSuccessful(true); result.setData(null); } return result; }  @Override SQSQueueResult processRequest(SQSRequestContext sqsRequestContext); void setWaitTimeSeconds(int waitTimeSeconds);  }