@Test public void testMessageDigestException() throws Exception { doThrow(new NoSuchAlgorithmException()).when(mockMessageDigestUtil).getMD5String(anyString()); QueueResult result = addRequestProcessor.processRequest( new SQSRequestContext(null, new QueueRequest().setData(Collections.singletonList("")), null)); Assert.assertFalse(result.isSuccessful()); }
@Override public SQSQueueResult processRequest(SQSRequestContext sqsRequestContext) { Map<String, String> idMD5Map = new HashMap<>(); Collection<String> data = sqsRequestContext.getQueueRequest().getData(); if(data == null) return new SQSQueueResult().setSuccessful(true); SendMessageBatchRequest sendMessageBatchRequest; try { sendMessageBatchRequest = new SendMessageBatchRequest() .withQueueUrl(sqsRequestContext.getQueueURL()) .withEntries(data.stream().map(item -> createSendMessageBatchRequestEntry(idMD5Map, item) ).collect(Collectors.toList()) ); } catch(RuntimeException e) { return new SQSQueueResult() .setSuccessful(false) .setMessage(e.getCause() == null ? e.getMessage() : e.getCause().getMessage()); } SendMessageBatchResult sendMessageBatchResult = sqsRequestContext.getSqsClient().sendMessageBatch(sendMessageBatchRequest); if(sendMessageBatchResult.getFailed() != null && sendMessageBatchResult.getFailed().size() > 0) { return new SQSQueueResult() .setSuccessful(false) .setMessage(String.format("Failure to persist [%1$s] messages", sendMessageBatchResult.getFailed().size())); } AtomicBoolean md5Errors = new AtomicBoolean(false); sendMessageBatchResult.getSuccessful().forEach( result -> { if(!validateResponse(idMD5Map.get(result.getId()), result.getMD5OfMessageBody())) { logger.error(String.format("MD5 of source body does not match result body: [%1$s][%2$s]", idMD5Map.get(result.getId()), result.getMD5OfMessageBody())); md5Errors.set(true); } } ); if(md5Errors.get()) { return new SQSQueueResult() .setSuccessful(false) .setMessage("MD5 mismatches when persisting to queue"); } return new SQSQueueResult() .setSuccessful(true); }
AddRequestProcessor implements SQSRequestProcessor { @Override public SQSQueueResult processRequest(SQSRequestContext sqsRequestContext) { Map<String, String> idMD5Map = new HashMap<>(); Collection<String> data = sqsRequestContext.getQueueRequest().getData(); if(data == null) return new SQSQueueResult().setSuccessful(true); SendMessageBatchRequest sendMessageBatchRequest; try { sendMessageBatchRequest = new SendMessageBatchRequest() .withQueueUrl(sqsRequestContext.getQueueURL()) .withEntries(data.stream().map(item -> createSendMessageBatchRequestEntry(idMD5Map, item) ).collect(Collectors.toList()) ); } catch(RuntimeException e) { return new SQSQueueResult() .setSuccessful(false) .setMessage(e.getCause() == null ? e.getMessage() : e.getCause().getMessage()); } SendMessageBatchResult sendMessageBatchResult = sqsRequestContext.getSqsClient().sendMessageBatch(sendMessageBatchRequest); if(sendMessageBatchResult.getFailed() != null && sendMessageBatchResult.getFailed().size() > 0) { return new SQSQueueResult() .setSuccessful(false) .setMessage(String.format("Failure to persist [%1$s] messages", sendMessageBatchResult.getFailed().size())); } AtomicBoolean md5Errors = new AtomicBoolean(false); sendMessageBatchResult.getSuccessful().forEach( result -> { if(!validateResponse(idMD5Map.get(result.getId()), result.getMD5OfMessageBody())) { logger.error(String.format("MD5 of source body does not match result body: [%1$s][%2$s]", idMD5Map.get(result.getId()), result.getMD5OfMessageBody())); md5Errors.set(true); } } ); if(md5Errors.get()) { return new SQSQueueResult() .setSuccessful(false) .setMessage("MD5 mismatches when persisting to queue"); } return new SQSQueueResult() .setSuccessful(true); } }
AddRequestProcessor implements SQSRequestProcessor { @Override public SQSQueueResult processRequest(SQSRequestContext sqsRequestContext) { Map<String, String> idMD5Map = new HashMap<>(); Collection<String> data = sqsRequestContext.getQueueRequest().getData(); if(data == null) return new SQSQueueResult().setSuccessful(true); SendMessageBatchRequest sendMessageBatchRequest; try { sendMessageBatchRequest = new SendMessageBatchRequest() .withQueueUrl(sqsRequestContext.getQueueURL()) .withEntries(data.stream().map(item -> createSendMessageBatchRequestEntry(idMD5Map, item) ).collect(Collectors.toList()) ); } catch(RuntimeException e) { return new SQSQueueResult() .setSuccessful(false) .setMessage(e.getCause() == null ? e.getMessage() : e.getCause().getMessage()); } SendMessageBatchResult sendMessageBatchResult = sqsRequestContext.getSqsClient().sendMessageBatch(sendMessageBatchRequest); if(sendMessageBatchResult.getFailed() != null && sendMessageBatchResult.getFailed().size() > 0) { return new SQSQueueResult() .setSuccessful(false) .setMessage(String.format("Failure to persist [%1$s] messages", sendMessageBatchResult.getFailed().size())); } AtomicBoolean md5Errors = new AtomicBoolean(false); sendMessageBatchResult.getSuccessful().forEach( result -> { if(!validateResponse(idMD5Map.get(result.getId()), result.getMD5OfMessageBody())) { logger.error(String.format("MD5 of source body does not match result body: [%1$s][%2$s]", idMD5Map.get(result.getId()), result.getMD5OfMessageBody())); md5Errors.set(true); } } ); if(md5Errors.get()) { return new SQSQueueResult() .setSuccessful(false) .setMessage("MD5 mismatches when persisting to queue"); } return new SQSQueueResult() .setSuccessful(true); } AddRequestProcessor(); }
AddRequestProcessor implements SQSRequestProcessor { @Override public SQSQueueResult processRequest(SQSRequestContext sqsRequestContext) { Map<String, String> idMD5Map = new HashMap<>(); Collection<String> data = sqsRequestContext.getQueueRequest().getData(); if(data == null) return new SQSQueueResult().setSuccessful(true); SendMessageBatchRequest sendMessageBatchRequest; try { sendMessageBatchRequest = new SendMessageBatchRequest() .withQueueUrl(sqsRequestContext.getQueueURL()) .withEntries(data.stream().map(item -> createSendMessageBatchRequestEntry(idMD5Map, item) ).collect(Collectors.toList()) ); } catch(RuntimeException e) { return new SQSQueueResult() .setSuccessful(false) .setMessage(e.getCause() == null ? e.getMessage() : e.getCause().getMessage()); } SendMessageBatchResult sendMessageBatchResult = sqsRequestContext.getSqsClient().sendMessageBatch(sendMessageBatchRequest); if(sendMessageBatchResult.getFailed() != null && sendMessageBatchResult.getFailed().size() > 0) { return new SQSQueueResult() .setSuccessful(false) .setMessage(String.format("Failure to persist [%1$s] messages", sendMessageBatchResult.getFailed().size())); } AtomicBoolean md5Errors = new AtomicBoolean(false); sendMessageBatchResult.getSuccessful().forEach( result -> { if(!validateResponse(idMD5Map.get(result.getId()), result.getMD5OfMessageBody())) { logger.error(String.format("MD5 of source body does not match result body: [%1$s][%2$s]", idMD5Map.get(result.getId()), result.getMD5OfMessageBody())); md5Errors.set(true); } } ); if(md5Errors.get()) { return new SQSQueueResult() .setSuccessful(false) .setMessage("MD5 mismatches when persisting to queue"); } return new SQSQueueResult() .setSuccessful(true); } AddRequestProcessor(); @Override SQSQueueResult processRequest(SQSRequestContext sqsRequestContext); void setMessageDigestUtil(MessageDigestUtil messageDigestUtil); void setIdGenerator(IdGenerator uuidGenerator); }
AddRequestProcessor implements SQSRequestProcessor { @Override public SQSQueueResult processRequest(SQSRequestContext sqsRequestContext) { Map<String, String> idMD5Map = new HashMap<>(); Collection<String> data = sqsRequestContext.getQueueRequest().getData(); if(data == null) return new SQSQueueResult().setSuccessful(true); SendMessageBatchRequest sendMessageBatchRequest; try { sendMessageBatchRequest = new SendMessageBatchRequest() .withQueueUrl(sqsRequestContext.getQueueURL()) .withEntries(data.stream().map(item -> createSendMessageBatchRequestEntry(idMD5Map, item) ).collect(Collectors.toList()) ); } catch(RuntimeException e) { return new SQSQueueResult() .setSuccessful(false) .setMessage(e.getCause() == null ? e.getMessage() : e.getCause().getMessage()); } SendMessageBatchResult sendMessageBatchResult = sqsRequestContext.getSqsClient().sendMessageBatch(sendMessageBatchRequest); if(sendMessageBatchResult.getFailed() != null && sendMessageBatchResult.getFailed().size() > 0) { return new SQSQueueResult() .setSuccessful(false) .setMessage(String.format("Failure to persist [%1$s] messages", sendMessageBatchResult.getFailed().size())); } AtomicBoolean md5Errors = new AtomicBoolean(false); sendMessageBatchResult.getSuccessful().forEach( result -> { if(!validateResponse(idMD5Map.get(result.getId()), result.getMD5OfMessageBody())) { logger.error(String.format("MD5 of source body does not match result body: [%1$s][%2$s]", idMD5Map.get(result.getId()), result.getMD5OfMessageBody())); md5Errors.set(true); } } ); if(md5Errors.get()) { return new SQSQueueResult() .setSuccessful(false) .setMessage("MD5 mismatches when persisting to queue"); } return new SQSQueueResult() .setSuccessful(true); } AddRequestProcessor(); @Override SQSQueueResult processRequest(SQSRequestContext sqsRequestContext); void setMessageDigestUtil(MessageDigestUtil messageDigestUtil); void setIdGenerator(IdGenerator uuidGenerator);  }