项目：aws-ski-resort
文件：QueueService.java
public List<QueueMessage> getMessages() {
ReceiveMessageRequest receiveMessageRequest = new ReceiveMessageRequest(_queueDns);
List<Message> messages = _sqs.receiveMessage(receiveMessageRequest).getMessages();
List<QueueMessage> deserializedMessages = new ArrayList<>();
for (Message message : messages) {
String body = message.getBody();
QueueMessage qm = _gson.fromJson(body,QueueMessage.class);
deserializedMessages.add(qm);
System.out.println("query time: " + qm.queryExecutionTime);
System.out.println("exec time: " + qm.totalExecutionTime);
System.out.println("Has ex: " + qm.hasException);
System.out.println("ex message: " + qm.exceptionMessage + "\n");
String receiptHandle = message.getReceiptHandle();
_sqs.deleteMessage(new DeleteMessageRequest(_queueDns,receiptHandle));
}
return deserializedMessages;
}
项目：graylog-plugin-s3
文件：S3SNSNotificationParser.java
public List<S3SNSNotification> parse(Message message) {
List<S3SNSNotification> notifications = Lists.newArrayList();
try {
SQSMessage envelope = om.readValue(message.getBody(),SQSMessage.class);
if (envelope.message == null) {
return Collections.emptyList();
}
S3EventNotification s3EventNotification = S3EventNotification.parseJson(envelope.message);
notifications.addAll(s3EventNotification.getRecords().stream().map(record -> new S3SNSNotification(
message.getReceiptHandle(),record.getS3().getBucket().getName(),record.getS3().getObject().getUrlDecodedKey()
)).collect(Collectors.toList()));
} catch (Exception e) {
LOG.error("Could not parse SNS notification: " + message.getBody(),e);
throw new RuntimeException("Could not parse SNS notification: " + message.getBody(),e);
}
return notifications;
}
项目：rxjava2-aws
文件：SqsTest.java
@Test(timeout = 5000)
public void testFirstCallToReceiveMessagesReturnsOneMessage() {
final AmazonSQSClient sqs = Mockito.mock(AmazonSQSClient.class);
final String queueName = "queue";
Mockito.when(sqs.getQueueUrl(queueName)).thenAnswer(x -> new GetQueueUrlResult().withQueueUrl(queueName));
Mockito.when(sqs.receiveMessage(Mockito.<ReceiveMessageRequest>any()))
.thenReturn(new ReceiveMessageResult().withMessages(new Message().withBody("body1")));
Sqs.queueName(queueName) //
.sqsFactory(() -> sqs) //
.messages() //
.map(m -> m.message()) //
.doOnError(Throwable::printStackTrace) //
.take(1) //
.test() //
.awaitDone(10,TimeUnit.SECONDS) //
.assertComplete() //
.assertValue("body1");
final InOrder inorder = Mockito.inOrder(sqs);
inorder.verify(sqs,Mockito.atLeastOnce()).getQueueUrl(queueName);
inorder.verify(sqs,Mockito.times(1)).receiveMessage(Mockito.<ReceiveMessageRequest>any());
inorder.verify(sqs,Mockito.times(1)).shutdown();
inorder.verifyNoMoreInteractions();
}
项目：rxjava2-aws
文件：SqsTest.java
@Test(timeout = 5000)
public void testFirstCallToReceiveMessagesReturnsOneMessageAndHonoursBackpressure() {
final AmazonSQSClient sqs = Mockito.mock(AmazonSQSClient.class);
final String queueName = "queue";
Mockito.when(sqs.getQueueUrl(queueName)).thenAnswer(x -> new GetQueueUrlResult().withQueueUrl(queueName));
Mockito.when(sqs.receiveMessage(Mockito.<ReceiveMessageRequest>any()))
.thenReturn(new ReceiveMessageResult().withMessages(new Message().withBody("body1")));
Sqs.queueName(queueName) //
.sqsFactory(() -> sqs) //
.messages() //
.map(m -> m.message()) //
.doOnError(Throwable::printStackTrace) //
.test(0) //
.requestMore(1) //
.assertValue("body1")//
.assertNotComplete() //
.cancel();
final InOrder inorder = Mockito.inOrder(sqs);
inorder.verify(sqs,Mockito.times(1)).shutdown();
inorder.verifyNoMoreInteractions();
}
项目：rxjava2-aws
文件：SqsTest.java
@Test(timeout = 5000)
public void testFirstCallToReceiveMessagesReturnsNoMessagesThenSecondCallReturnsTwoMessages() {
final AmazonSQSClient sqs = Mockito.mock(AmazonSQSClient.class);
final String queueName = "queue";
Mockito.when(sqs.getQueueUrl(queueName)).thenAnswer(x -> new GetQueueUrlResult().withQueueUrl(queueName));
Mockito.when(sqs.receiveMessage(Mockito.<ReceiveMessageRequest>any())).thenReturn(new ReceiveMessageResult())
.thenReturn(new ReceiveMessageResult().withMessages(new Message().withBody("body1"),new Message().withBody("body2")));
Sqs.queueName(queueName) //
.sqsFactory(() -> sqs) //
.messages() //
.map(m -> m.message()) //
.doOnError(Throwable::printStackTrace) //
.take(2) //
.test() //
.awaitDone(10,TimeUnit.SECONDS) //
.assertComplete() //
.assertValues("body1","body2");
final InOrder inorder = Mockito.inOrder(sqs);
inorder.verify(sqs,Mockito.times(2)).receiveMessage(Mockito.<ReceiveMessageRequest>any());
inorder.verify(sqs,Mockito.times(1)).shutdown();
inorder.verifyNoMoreInteractions();
}
项目：aws-codecommit-trigger-plugin
文件：CodeCommitMessageParser.java
@Override
public List<Event> parseMessage(final Message message) {
List<Event> events = Collections.emptyList();
try {
MessageBody body = gson.fromJson(message.getBody(),MessageBody.class);
final String json = body.getMessage();
log.debug("Parse message %s",json);
if (StringUtils.isEmpty(json)) {
log.warning("Message contains no text");
return Collections.emptyList();
}
if (!json.startsWith("{") || !json.endsWith("}")) {
log.warning("Message text is no JSON");
return Collections.emptyList();
}
events = this.parseRecords(json);
} catch (final com.google.gson.JsonSyntaxException e) {
log.error("JSON syntax exception,cannot parse message: %s",e);
}
return events;
}
项目：aws-codecommit-trigger-plugin
文件：SQSTrigger.java
private boolean handleMessage(final Message message) {
log.debug("Parse and do match against events,message: %s",this.job,message.getBody());
final MessageParser parser = this.messageParserFactory.createParser(message);
final List<Event> events = parser.parseMessage(message);
boolean matched = this.eventTriggerMatcher.matches(events,this.sqsJob);
String messageId = com.ribose.jenkins.plugin.awscodecommittrigger.utils.StringUtils.getMessageId(message);
log.info("Any event matched? %s. Message: %s",matched,messageId);
if (matched) {
log.debug("Hurray! Execute it",this.job);
//TODO use java8 lambda for this loop
List<String> userarns = new ArrayList<>();
for (Event event : events) {
userarns.add(event.getUser());
}
this.execute(message,userarns);
return true;
}
return false;
}
项目：RekognitionS3Batch
文件：Processor.java
private void processTask(Message message) {
String path = message.getBody();
PathSplit pathComp = new PathSplit(path);
String bucket = pathComp.bucket;
String key = pathComp.key;
Logger.Info("Processing %s %s",bucket,key);
// Rekognition: Detect Labels from S3 object
DetectLabelsRequest req = new DetectLabelsRequest()
.withImage(new Image().withS3Object(new S3Object().withBucket(bucket).withName(key)))
.withMinConfidence(minConfidence);
DetectLabelsResult result;
result = rek.detectLabels(req);
List<Label> labels = result.getLabels();
Logger.Debug("In %s,found: %s",key,labels);
// Process downstream actions:
for (LabelProcessor processor : processors) {
processor.process(labels,path);
}
}
项目：emodb
文件：SQSScanWorkflow.java
@Override
public List<ScanRangeTask> claimScanRangeTasks(int max,Duration ttl) {
if (max == 0) {
return ImmutableList.of();
}
List<Message> messages = _sqs.receiveMessage(new ReceiveMessageRequest()
.withQueueUrl(getQueueUrl(_pendingScanRangeQueue))
.withMaxNumberOfMessages(Math.min(max,10)) // SQS cannot claim more than 10 messages
.withVisibilityTimeout(toSeconds(ttl))
).getMessages();
return FluentIterable.from(messages)
.transform(new Function<Message,ScanRangeTask>() {
@Override
public ScanRangeTask apply(Message message) {
QueueScanRangeTask task = JsonHelper.fromJson(message.getBody(),QueueScanRangeTask.class);
task.setMessageId(message.getReceiptHandle());
return task;
}
})
.toList();
}
项目：emodb
文件：SQSScanWorkflow.java
@Override
public List<ScanRangeComplete> claimCompleteScanRanges(Duration ttl) {
List<Message> messages = _sqs.receiveMessage(new ReceiveMessageRequest()
.withQueueUrl(getQueueUrl(_completeScanRangeQueue))
.withMaxNumberOfMessages(10)
.withVisibilityTimeout(toSeconds(ttl))
).getMessages();
return FluentIterable.from(messages)
.transform(new Function<Message,ScanRangeComplete>() {
@Override
public ScanRangeComplete apply(Message message) {
QueueScanRangeComplete completion = JsonHelper.fromJson(message.getBody(),QueueScanRangeComplete.class);
completion.setMessageId(message.getReceiptHandle());
return completion;
}
})
.toList();
}
项目：thingsboard
文件：SqsDemoClient.java
private void pollMessages(AmazonSQS sqs) {
log.info("Polling messages");
while (true) {
List<Message> messages = sqs.receiveMessage(QUEUE_URL).getMessages();
messages.forEach(m -> {
log.info("Message Received: " + m.getBody());
System.out.println(m.getBody());
DeleteMessageRequest deleteMessageRequest = new DeleteMessageRequest(QUEUE_URL,m.getReceiptHandle());
sqs.deleteMessage(deleteMessageRequest);
});
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
e.printStackTrace();
}
}
}
项目：rxjava-aws
文件：SqsTest.java
@Test(timeout = 5000)
public void testFirstCallToReceiveMessagesReturnsOneMessage() {
AmazonSQSClient sqs = Mockito.mock(AmazonSQSClient.class);
String queueName = "queue";
Mockito.when(sqs.getQueueUrl(queueName)).thenAnswer(x -> new GetQueueUrlResult().withQueueUrl(queueName));
Mockito.when(sqs.receiveMessage(Mockito.<ReceiveMessageRequest>any()))
.thenReturn(new ReceiveMessageResult().withMessages(new Message().withBody("body1")));
Sqs.queueName(queueName) //
.sqsFactory(() -> sqs) //
.messages() //
.map(m -> m.message()) //
.doOnError(Throwable::printStackTrace) //
.take(1) //
.to(test()) //
.awaitTerminalEvent() //
.assertCompleted() //
.assertValue("body1");
InOrder inorder = Mockito.inOrder(sqs);
inorder.verify(sqs,Mockito.times(1)).shutdown();
inorder.verifyNoMoreInteractions();
}
项目：rxjava-aws
文件：SqsTest.java
@Test(timeout = 5000)
public void testFirstCallToReceiveMessagesReturnsOneMessageAndHonoursBackpressure() {
AmazonSQSClient sqs = Mockito.mock(AmazonSQSClient.class);
String queueName = "queue";
Mockito.when(sqs.getQueueUrl(queueName)).thenAnswer(x -> new GetQueueUrlResult().withQueueUrl(queueName));
Mockito.when(sqs.receiveMessage(Mockito.<ReceiveMessageRequest>any()))
.thenReturn(new ReceiveMessageResult().withMessages(new Message().withBody("body1")));
Sqs.queueName(queueName) //
.sqsFactory(() -> sqs) //
.messages() //
.map(m -> m.message()) //
.doOnError(Throwable::printStackTrace) //
.to(testWithRequest(0)) //
.requestMore(1) //
.assertValue("body1")//
.assertNotCompleted() //
.unsubscribe();
InOrder inorder = Mockito.inOrder(sqs);
inorder.verify(sqs,Mockito.times(1)).shutdown();
inorder.verifyNoMoreInteractions();
}
项目：rxjava-aws
文件：SqsTest.java
@Test(timeout = 5000)
public void testFirstCallToReceiveMessagesReturnsNoMessagesThenSecondCallReturnsTwoMessages() {
AmazonSQSClient sqs = Mockito.mock(AmazonSQSClient.class);
String queueName = "queue";
Mockito.when(sqs.getQueueUrl(queueName)).thenAnswer(x -> new GetQueueUrlResult().withQueueUrl(queueName));
Mockito.when(sqs.receiveMessage(Mockito.<ReceiveMessageRequest>any())).thenReturn(new ReceiveMessageResult())
.thenReturn(new ReceiveMessageResult().withMessages(new Message().withBody("body1"),new Message().withBody("body2")));
Sqs.queueName(queueName) //
.sqsFactory(() -> sqs) //
.messages() //
.map(m -> m.message()) //
.doOnError(Throwable::printStackTrace) //
.take(2) //
.to(test()) //
.awaitTerminalEvent() //
.assertCompleted() //
.assertValues("body1","body2");
InOrder inorder = Mockito.inOrder(sqs);
inorder.verify(sqs,Mockito.times(1)).shutdown();
inorder.verifyNoMoreInteractions();
}
项目：Camel
文件：SqsDoesNotExtendMessageVisibilityTest.java
@Test
public void defaultsToDisabled() throws Exception {
this.mock.expectedMessageCount(1);
this.mock.whenAnyExchangeReceived(new Processor() {
@Override
public void process(Exchange exchange) throws Exception {
// Simulate message that takes a while to receive.
Thread.sleep(TIMEOUT * 1500L); // 150% of TIMEOUT.
}
});
Message message = new Message();
message.setBody("Message 1");
message.setMD5OfBody("6a1559560f67c5e7a7d5d838bf0272ee");
message.setMessageId("f6fb6f99-5eb2-4be4-9b15-144774141458");
message.setReceiptHandle(RECEIPT_HANDLE);
this.clientMock.messages.add(message);
assertMockEndpointsSatisfied(); // Wait for message to arrive.
assertTrue("Expected no changeMessageVisibility requests.",this.clientMock.changeMessageVisibilityRequests.size() == 0);
}
项目：Camel
文件：SqsBatchConsumerTest.java
@Override
protected JndiRegistry createRegistry() throws Exception {
JndiRegistry registry = super.createRegistry();
AmazonSQSClientMock clientMock = new AmazonSQSClientMock();
// add 6 messages,one more we will poll
for (int counter = 0; counter < 6; counter++) {
Message message = new Message();
message.setBody("Message " + counter);
message.setMD5OfBody("6a1559560f67c5e7a7d5d838bf0272ee");
message.setMessageId("f6fb6f99-5eb2-4be4-9b15-144774141458");
message.setReceiptHandle("0NNAq8PwvXsyZkR6yu4nQ07FGxNmOBWi5");
clientMock.messages.add(message);
}
registry.bind("amazonSQSClient",clientMock);
return registry;
}
项目：Camel
文件：AmazonSQSClientMock.java
@Override
public SendMessageResult sendMessage(SendMessageRequest sendMessageRequest) throws AmazonServiceException,AmazonClientException {
Message message = new Message();
message.setBody(sendMessageRequest.getMessageBody());
message.setMD5OfBody("6a1559560f67c5e7a7d5d838bf0272ee");
message.setMessageId("f6fb6f99-5eb2-4be4-9b15-144774141458");
message.setReceiptHandle("0NNAq8PwvXsyZkR6yu4nQ07FGxNmOBWi5zC9+4QMqJZ0DJ3gVOmjI2Gh/oFnb0IeJqy5Zc8kH4JX7GVpfjcEDjaAPSeOkXQZRcaBqt"
+ "4lOtyfj0kcclVV/zS7aenhfhX5Ixfgz/rHhsJwtCPPvTAdgQFGYrqaHly+etJiawiNPVc=");
synchronized (messages) {
messages.add(message);
}
SendMessageResult result = new SendMessageResult();
result.setMessageId("f6fb6f99-5eb2-4be4-9b15-144774141458");
result.setMD5OfMessageBody("6a1559560f67c5e7a7d5d838bf0272ee");
return result;
}
项目：Camel
文件：AmazonSQSClientMock.java
@Override
public ReceiveMessageResult receiveMessage(ReceiveMessageRequest receiveMessageRequest) throws AmazonServiceException,AmazonClientException {
Integer maxNumberOfMessages = receiveMessageRequest.getMaxNumberOfMessages() != null ? receiveMessageRequest.getMaxNumberOfMessages() : Integer.MAX_VALUE;
ReceiveMessageResult result = new ReceiveMessageResult();
Collection<Message> resultMessages = new ArrayList<Message>();
synchronized (messages) {
int fetchSize = 0;
for (Iterator<Message> iterator = messages.iterator(); iterator.hasNext() && fetchSize < maxNumberOfMessages; fetchSize++) {
Message rc = iterator.next();
resultMessages.add(rc);
iterator.remove();
scheduleCancelInflight(receiveMessageRequest.getQueueUrl(),rc);
}
}
result.setMessages(resultMessages);
return result;
}
项目：Camel
文件：AmazonSQSClientMock.java
private void scheduleCancelInflight(final String queueUrl,final Message message) {
if (scheduler != null) {
int visibility = getVisibilityForQueue(queueUrl);
if (visibility > 0) {
ScheduledFuture task = scheduler.schedule(new Runnable() {
@Override
public void run() {
synchronized (messages) {
// put it back!
messages.add(message);
}
}
},visibility,TimeUnit.SECONDS);
inFlight.put(message.getReceiptHandle(),task);
}
}
}
项目：s3_video
文件：AWSAdapter.java
public List<JobStatusNotification> pollMessageFromQueueByJobId(String queueUrl,String jobId) {
ReceiveMessageRequest receiveMessageRequest = new ReceiveMessageRequest()
.withQueueUrl(queueUrl)
.withMaxNumberOfMessages(MAX_NUMBER_OF_MESSAGES)
.withVisibilityTimeout(VISIBILITY_TIMEOUT)
.withWaitTimeSeconds(WAIT_TIME_SECONDS);
List<JobStatusNotification> jobStatusNotifications = new ArrayList<>();
for (Message message : sqsClient.receiveMessage(receiveMessageRequest).getMessages()) {
try {
JobStatusNotification jobStatusNotification = parseMessage(message.getBody());
if (jobStatusNotification.getJobId().equalsIgnoreCase(jobId)) {
jobStatusNotifications.add(jobStatusNotification);
sqsClient.deleteMessage(new DeleteMessageRequest().withQueueUrl(queueUrl).withReceiptHandle(message.getReceiptHandle()));
}
} catch (IOException e) {
logger.error(e.getMessage(),e);
}
}
return jobStatusNotifications;
}
项目：distributed-image-classification
文件：Queue.java
@SuppressWarnings("unchecked")
public T peekMessage(int waitFor) throws Exception {
// Receive messages
logger.info("Trying to recieve message from: " + _queueName);
ReceiveMessageRequest receiveMessageRequest = new ReceiveMessageRequest(_queueURL);
receiveMessageRequest.setMaxNumberOfMessages(1);
receiveMessageRequest.setWaitTimeSeconds(waitFor);
List<Message> messages = _sqs.receiveMessage(receiveMessageRequest).getMessages();
for (Message message : messages) {
logger.info(" Got Message");
logger.info(" Body: " + message.getBody());
logger.info(" Handle: " + message.getReceiptHandle());
_lastMessage = message;
GenericMessage msg = GenericMessage.fromXML(message.getBody());
if (!msg.type.equals(_msgClass.getName()))
throw new Exception("Invalid message type recieved.");
return (T) msg.body;
}
return null;
}
项目：pipeline
文件：AmazonSQSIterator.java
/**
* Delete a message from the SQS queue
*
* @param messageHandle
* Message handle to delete
* @return true if the delete was successful,otherwise false
*/
public boolean deleteMessage(@Nullable final Message message) {
if (message == null) {
return false;
}
try {
LOGGER.debug("Deleting message from SQS: {}",message.getMessageId());
deleteRequests.inc();
sqs.deleteMessage(queueUrl,message.getReceiptHandle());
return true;
} catch (Exception e) {
LOGGER.error("Failed to delete message: " + message.getMessageId(),e);
}
return false;
}
项目：dropwizard-sqs-bundle
文件：SqsBundle.java
public <T> void registerReceiver(String queueName,SqsReceiver<T> receiver) {
Optional<String> queueUrl = getUrlForQueue(queueName);
if (queueUrl.isPresent()) {
SqsReceiverHandler<T> handler = new SqsReceiverHandler<>(
sqs,queueUrl.get(),receiver,new SqsBaseExceptionHandler() { // not replaced with lambda because jacoco fails with lambdas
@Override
public boolean onException(Message message,Exception exception) {
LOGGER.error("Error processing received message - acknowledging it anyway");
return true;
}
}
);
internalRegisterReceiver(queueName,handler);
}
else {
LOGGER.error("Cannot register receiver for queue name : " + queueName);
}
}
项目：dropwizard-sqs-bundle
文件：SqsReceiverHandlerTest.java
@Test
public void messageShouldBeProcessedAfterBeingConsumed() throws Exception {
//GIVEN
ReceiveMessageResult receiveMessageResult = new ReceiveMessageResult();
Message message1 = new Message()
.withMessageId("aaaa-bbbb-cccc-dddd-eeee")
.withBody("Sample test message");
Message message2 = new Message()
.withMessageId("ffff-gggg-hhhh-iiii-jjjj")
.withBody("Another sample test message");
receiveMessageResult.setMessages(Lists.newArrayList(message1,message2));
when(sqs.receiveMessage((ReceiveMessageRequest) anyObject())).thenReturn(receiveMessageResult,new ReceiveMessageResult());
//WHEN
receiverHandler.start();
//THEN
Thread.sleep(1000);
verify(receiver,times(2)).receive(any());
verify(receiver,times(1)).receive(message1);
verify(receiver,times(1)).receive(message2);
}
项目：aws-java-sdk-stubs
文件：AmazonSQSStubTest.java
@Test
public void sendAndReceiveMessage() {
final String queueName = "bizo";
final String messageBody = "hi everybody";
final CreateQueueRequest createQueueRequest = new CreateQueueRequest().withQueueName(queueName);
sqs.createQueue(createQueueRequest);
final GetQueueUrlRequest getQueueUrlRequest = new GetQueueUrlRequest().withQueueName(queueName);
final GetQueueUrlResult getQueueUrlResult = sqs.getQueueUrl(getQueueUrlRequest);
final String queueUrl = getQueueUrlResult.getQueueUrl();
final SendMessageRequest sendMessageRequest =
new SendMessageRequest().withQueueUrl(queueUrl).withMessageBody(messageBody);
sqs.sendMessage(sendMessageRequest);
final int maxNumberOfMessages = 10;
final ReceiveMessageRequest receiveMessageRequest =
new ReceiveMessageRequest().withQueueUrl(queueUrl).withMaxNumberOfMessages(maxNumberOfMessages);
final ReceiveMessageResult receiveMessageResult = sqs.receiveMessage(receiveMessageRequest);
final List<Message> messages = receiveMessageResult.getMessages();
assertThat(messages.size(),equalTo(1));
assertThat(messages.get(0).getBody(),equalTo(messageBody));
}
项目：graylog-plugin-aws
文件：CloudtrailSNSNotificationParserTest.java
@Test
public void testParse() throws Exception {
final Message message = new Message()
.withBody("{\n" +
" \"Type\" : \"Notification\",\n" +
" \"MessageId\" : \"55508fe9-870b-590c-960f-c34960b669f0\",\n" +
" \"TopicArn\" : \"arn:aws:sns:eu-west-1:459220251735:cloudtrail-write\",\n" +
" \"Message\" : \"{\\\"s3Bucket\\\":\\\"cloudtrailbucket\\\",\\\"s3ObjectKey\\\":[\\\"example/AWSLogs/459220251735/CloudTrail/eu-west-1/2014/09/27/459220251735_CloudTrail_eu-west-1_20140927T1625Z_UPwzr7ft2mf0Q1SS.json.gz\\\"]}\",\n" +
" \"Timestamp\" : \"2014-09-27T16:27:41.258Z\",\n" +
" \"SignatureVersion\" : \"1\",\n" +
" \"Signature\" : \"O05joR97NvGHqMJQwsSNXzeSHrtbLqbRcqsXB7xmqARyaCGXjaVh2duwTUL93s4YvoNENnOEMzkILKI5PwmQQPha5/cmj6FSjblwRMMga6Xzf6cMnurT9TphQO7z35foHG49IejW05IkzIwD/DW0GvafJLah+fQI3EFySnShzXLFESGQuumdS8bxnM5r96ne8t+MEAHfBCVyQ/QrduO9tTtfXAz6OeWg1IEwV3TeZ5c5SS5vRxxhsD4hOJSmXAUM99CeQfcG9s7saBcvyyGPZrhPEh8S1uhiTmLvr6h1voM9vgiCbCCUujExvg+bnqsXWTZBmnatF1iOyxFfYcZ6kw==\",\n" +
" \"SigningCertURL\" : \"https://sns.eu-west-1.amazonaws.com/SimpleNotificationService-d6d679a1d18e95c2f9ffcf11f4f9e198.pem\",\n" +
" \"UnsubscribeURL\" : \"https://sns.eu-west-1.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn:aws:sns:eu-west-1:459220251735:cloudtrail-write:9a3a4e76-4173-4c8c-b488-0126315ba643\"\n" +
"}");
CloudtrailSNSNotificationParser parser = new CloudtrailSNSNotificationParser(objectMapper);
List<CloudtrailSNSNotification> notifications = parser.parse(message);
assertEquals(1,notifications.size());
CloudtrailSNSNotification notification = notifications.get(0);
assertEquals(notification.getS3Bucket(),"cloudtrailbucket");
assertEquals(notification.getS3ObjectKey(),"example/AWSLogs/459220251735/CloudTrail/eu-west-1/2014/09/27/459220251735_CloudTrail_eu-west-1_20140927T1625Z_UPwzr7ft2mf0Q1SS.json.gz");
}
项目：graylog-plugin-aws
文件：CloudtrailSNSNotificationParserTest.java
@Test
public void issue_44() throws Exception {
// https://github.com/Graylog2/graylog-plugin-aws/issues/44
final Message message = new Message()
.withBody("{\n" +
" \"Type\" : \"Notification\",\n" +
" \"MessageId\" : \"5b0a73e6-a4f8-11e7-8dfb-8f76310a10a8\",\n" +
" \"TopicArn\" : \"arn:aws:sns:eu-west-1:123456789012:cloudtrail-log-write\",\n" +
" \"Subject\" : \"[AWS Config:eu-west-1] AWS::RDS::DBSnapshot rds:instance-2017-09-03-23-11 Dele...\",\n" +
" \"Message\" : \"{\\\"configurationItemDiff\\\":{\\\"changedProperties\\\":{\\\"Relationships.0\\\":{\\\"previousValue\\\":{\\\"resourceId\\\":\\\"vpc-12345678\\\",\\\"resourceName\\\":null,\\\"resourceType\\\":\\\"AWS::EC2::VPC\\\",\\\"name\\\":\\\"Is associated with Vpc\\\"},\\\"updatedValue\\\":null,\\\"changeType\\\":\\\"DELETE\\\"},\\\"SupplementaryConfiguration.Tags\\\":{\\\"previousValue\\\":[],\\\"SupplementaryConfiguration.DBSnapshotAttributes\\\":{\\\"previousValue\\\":[{\\\"attributeName\\\":\\\"restore\\\",\\\"attributeValues\\\":[]}],\\\"Configuration\\\":{\\\"previousValue\\\":{\\\"dBSnapshotIdentifier\\\":\\\"rds:instance-2017-09-03-23-11\\\",\\\"dBInstanceIdentifier\\\":\\\"instance\\\",\\\"snapshotCreateTime\\\":\\\"2017-09-03T23:11:38.218Z\\\",\\\"engine\\\":\\\"mysql\\\",\\\"allocatedStorage\\\":200,\\\"status\\\":\\\"available\\\",\\\"port\\\":3306,\\\"availabilityZone\\\":\\\"eu-west-1b\\\",\\\"vpcId\\\":\\\"vpc-12345678\\\",\\\"instanceCreateTime\\\":\\\"2015-04-09T07:08:07.476Z\\\",\\\"masterUsername\\\":\\\"root\\\",\\\"engineVersion\\\":\\\"5.6.34\\\",\\\"licenseModel\\\":\\\"general-public-license\\\",\\\"snapshotType\\\":\\\"automated\\\",\\\"iops\\\":null,\\\"optionGroupName\\\":\\\"default:mysql-5-6\\\",\\\"percentProgress\\\":100,\\\"sourceRegion\\\":null,\\\"sourceDBSnapshotIdentifier\\\":null,\\\"storageType\\\":\\\"standard\\\",\\\"tdeCredentialArn\\\":null,\\\"encrypted\\\":false,\\\"kmsKeyId\\\":null,\\\"dBSnapshotArn\\\":\\\"arn:aws:rds:eu-west-1:123456789012:snapshot:rds:instance-2017-09-03-23-11\\\",\\\"timezone\\\":null,\\\"iAMDatabaseAuthenticationEnabled\\\":false},\\\"changeType\\\":\\\"DELETE\\\"}},\\\"configurationItem\\\":{\\\"relatedEvents\\\":[],\\\"relationships\\\":[],\\\"configuration\\\":null,\\\"supplementaryConfiguration\\\":{},\\\"tags\\\":{},\\\"configurationItemVersion\\\":\\\"1.2\\\",\\\"configurationItemCaptureTime\\\":\\\"2017-09-28T19:54:47.815Z\\\",\\\"configurationStateId\\\":1234567890123,\\\"awsAccountId\\\":\\\"123456789012\\\",\\\"configurationItemStatus\\\":\\\"ResourceDeleted\\\",\\\"resourceType\\\":\\\"AWS::RDS::DBSnapshot\\\",\\\"resourceId\\\":\\\"rds:instance-2017-09-03-23-11\\\",\\\"resourceName\\\":\\\"rds:instance-2017-09-03-23-11\\\",\\\"ARN\\\":\\\"arn:aws:rds:eu-west-1:123456789012:snapshot:rds:instance-2017-09-03-23-11\\\",\\\"awsRegion\\\":\\\"eu-west-1\\\",\\\"availabilityZone\\\":null,\\\"configurationStateMd5Hash\\\":\\\"b026324c6904b2a9cb4b88d6d61c81d1\\\",\\\"resourceCreationTime\\\":null},\\\"notificationCreationTime\\\":\\\"2017-09-28T19:54:48.311Z\\\",\\\"messageType\\\":\\\"ConfigurationItemChangeNotification\\\",\\\"recordVersion\\\":\\\"1.2\\\"}\",\n" +
" \"Timestamp\" : \"2017-09-28T19:54:58.543Z\",\n" +
" \"Signature\" : \"...\",\n" +
" \"SigningCertURL\" : \"https://sns.eu-west-1.amazonaws.com/SimpleNotificationService-....pem\",\n" +
" \"UnsubscribeURL\" : \"https://sns.eu-west-1.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn:aws:sns:eu-west-1:123456789012:cloudtrail-log-write:5b0a73e6-a4f8-11e7-8dfb-8f76310a10a8\"\n" +
"}");
CloudtrailSNSNotificationParser parser = new CloudtrailSNSNotificationParser(objectMapper);
List<CloudtrailSNSNotification> notifications = parser.parse(message);
assertTrue(notifications.isEmpty());
}
项目：aws-cloudtrail-processing-library
文件：CloudTrailSourceSerializer.java
/**
* As long as there is at least one CloudTrail log object:
* <p>
* <li>Add the CloudTrail log object key to the list.</li>
* <li>Add <code>accountId</code> extracted from log object key to <code>sqsMessage</code>.</li>
* <li>Add {@link SourceType#CloudTrailLog} to the <code>sqsMessage</code>.</li>
* </p>
*
* If there is no CloudTrail log object and it is a valid CloudTrail message,CPL adds only {@link SourceType#Other}
* to the <code>sqsMessage</code>.
*
*/
private void addCloudTrailLogsAndMessageAttributes(Message sqsMessage,List<CloudTrailLog> cloudTrailLogs,JsonNode messageNode) throws IOException {
SourceType sourceType = SourceType.Other;
String bucketName = messageNode.get(S3_BUCKET_NAME).textValue();
List<String> objectKeys = mapper.readValue(messageNode.get(S3_OBJECT_KEY).traverse(),new TypeReference<List<String>>() {});
for (String objectKey: objectKeys) {
SourceType currSourceType = sourceIdentifier.identify(objectKey);
if (currSourceType == SourceType.CloudTrailLog) {
cloudTrailLogs.add(new CloudTrailLog(bucketName,objectKey));
sourceType = currSourceType;
LibraryUtils.setMessageAccountId(sqsMessage,objectKey);
}
}
sqsMessage.addAttributesEntry(SourceAttributeKeys.SOURCE_TYPE.getAttributeKey(),sourceType.name());
}
项目：aws-cloudtrail-processing-library
文件：S3SourceSerializer.java
/**
* As long as there is at least one CloudTrail log object:
* <p>
* <li>Add the CloudTrail log object key to the list.</li>
* <li>Add <code>accountId</code> extracted from log object key to <code>sqsMessage</code>.</li>
* <li>Add {@link SourceType#CloudTrailLog} to the <code>sqsMessage</code>.</li>
* </p>
*
* If there is no CloudTrail log object and it is a valid S3 message,JsonNode s3RecordsNode,List<CloudTrailLog> cloudTrailLogs) {
SourceType sourceType = SourceType.Other;
for (JsonNode s3Record: s3RecordsNode) {
String bucketName = s3Record.at(S3_BUCKET_NAME).textValue();
String objectKey = s3Record.at(S3_OBJECT_KEY).textValue();
String eventName = s3Record.get(EVENT_NAME).textValue();
SourceType currSourceType = sourceIdentifier.identifyWithEventName(objectKey,eventName);
if (currSourceType == SourceType.CloudTrailLog) {
cloudTrailLogs.add(new CloudTrailLog(bucketName,sourceType.name());
}
项目：support
文件：SQS.java
public static String popFrom(String name) {
try {
String queueUrl = getConnection().createQueue(
new CreateQueueRequest(name)).getQueueUrl();
ReceiveMessageRequest receiveMessageRequest = new ReceiveMessageRequest(
queueUrl);
receiveMessageRequest.setMaxNumberOfMessages(1);
if (null != receiveMessageRequest) {
List<Message> messages = getConnection().receiveMessage(
receiveMessageRequest).getMessages();
if (messages.size() > 0) {
String messageRecieptHandle = messages.get(0)
.getReceiptHandle();
getConnection().deleteMessage(
new DeleteMessageRequest(receiveMessageRequest
.getQueueUrl(),messageRecieptHandle));
return messages.get(0).getBody();
}
}
} catch (Exception e) {
e.printStackTrace();
}
return null;
}
项目：amazon-sqs-java-messaging-lib
文件：SQSSessionCallbackSchedulerTest.java
private SQSMessageConsumerPrefetch.MessageManager createFifoMessageManager(String queueUrl,String groupId,String messageId,String receiptHandle) throws JMSException {
Message message = new Message();
message.setBody("body");
message.setMessageId(messageId);
message.setReceiptHandle(receiptHandle);
Map<String,String> attributes = new HashMap<String,String>();
attributes.put(SQSMessagingClientConstants.SEQUENCE_NUMBER,"728374687246872364");
attributes.put(SQSMessagingClientConstants.MESSAGE_DEDUPLICATION_ID,messageId);
attributes.put(SQSMessagingClientConstants.MESSAGE_GROUP_ID,groupId);
attributes.put(SQSMessagingClientConstants.APPROXIMATE_RECEIVE_COUNT,"0");
message.setAttributes(attributes);
SQSMessage sqsMessage = new SQSTextMessage(acknowledger,queueUrl,message);
PrefetchManager prefetchManager = mock(PrefetchManager.class);
when(prefetchManager.getMessageConsumer())
.thenReturn(consumer);
SQSMessageConsumerPrefetch.MessageManager msgManager = new SQSMessageConsumerPrefetch.MessageManager(prefetchManager,sqsMessage);
return msgManager;
}
项目：izettle-toolbox
文件：QueueProcessor.java
public static MessageQueueProcessor createQueueProcessor(
AmazonSQS amazonSQS,String name,String queueUrl,String deadLetterQueueUrl,MessageHandler<Message> messageHandler
) {
return new QueueProcessor(
name,deadLetterQueueUrl,amazonSQS,messageHandler,null
);
}
项目：izettle-toolbox
文件：QueueProcessor.java
public static MessageQueueProcessor createQueueProcessor(
AmazonSQS amazonSQS,MessageHandler<Message> messageHandler,ExecutorService executorService
) {
return new QueueProcessor(
name,executorService
);
}
项目：izettle-toolbox
文件：QueueProcessorTest.java
@Test
public void shouldPassAllPolledMessagesToSpecifiedHandler() throws Exception {
// Arrange
Message msg1 = createMessage("msg1");
Message msg2 = createMessage("msg2");
receivedMessages.add(msg1);
receivedMessages.add(msg2);
// Act
queueProcessor.poll();
// Assert
verify(mockHandler).handle(msg1);
verify(mockHandler).handle(msg2);
verifyNoMoreInteractions(mockHandler);
}
项目：izettle-toolbox
文件：QueueServicePollerTest.java
@Test
public void pollAndDeleteMessageShouldWork() throws Exception {
ReceiveMessageResult receiveMessageResult = mock(ReceiveMessageResult.class);
Message message = mock(Message.class);
when(message.getBody()).thenReturn("{}");
when(receiveMessageResult.getMessages()).thenReturn(Arrays.asList(message));
when(mockAmazonSQS.receiveMessage(any(ReceiveMessageRequest.class))).thenReturn(receiveMessageResult);
List<PolledMessage<TestMessage>> receivedMessages1 = queueServicePoller.poll();
assertThat(receivedMessages1).hasSize(1);
when(mockAmazonSQS.receiveMessage(any(ReceiveMessageRequest.class))).thenReturn(mock(ReceiveMessageResult.class));
queueServicePoller.delete(receivedMessages1.get(0));
List<PolledMessage<TestMessage>> receivedMessages2 = queueServicePoller.poll();
assertThat(receivedMessages2).isEmpty();
}
项目：izettle-toolbox
文件：QueueServicePollerTest.java
@Test
public void deleteBatchMessagesShouldWork() throws Exception {
ReceiveMessageResult receiveMessageResult = mock(ReceiveMessageResult.class);
Message message = mock(Message.class);
when(message.getBody()).thenReturn("{}");
when(receiveMessageResult.getMessages()).thenReturn(Arrays.asList(message,message));
when(mockAmazonSQS.receiveMessage(any(ReceiveMessageRequest.class))).thenReturn(receiveMessageResult);
List<PolledMessage<TestMessage>> receivedMessages1 = queueServicePoller.poll();
assertEquals(2,receivedMessages1.size());
when(mockAmazonSQS.receiveMessage(any(ReceiveMessageRequest.class))).thenReturn(mock(ReceiveMessageResult.class));
queueServicePoller.delete(receivedMessages1.get(0));
queueServicePoller.delete(receivedMessages1.get(1));
List<PolledMessage<TestMessage>> receivedMessages2 = queueServicePoller.poll();
assertEquals(0,receivedMessages2.size());
}
项目：awslocal
文件：DirectorySQS.java
@Override
public ReceiveMessageResult receiveMessage(ReceiveMessageRequest receiveMessageRequest) throws AmazonClientException {
DirectorySQSQueue queue = getQueueFromUrl(receiveMessageRequest.getQueueUrl(),false);
//make sure we have a default for max number of messages.
int maxNumberOfMessages = Objects.firstNonNull(receiveMessageRequest.getMaxNumberOfMessages(),10); //10 is amazon spec default
//and a default visibility timeout
int visibilityTimeout = Objects.firstNonNull(receiveMessageRequest.getVisibilityTimeout(),_defaultVisibilitySeconds);
//also a wait time
int waitTime = Objects.firstNonNull(receiveMessageRequest.getWaitTimeSeconds(),0);
if (waitTime < 0 || waitTime > 20) {
throw new AmazonServiceException("wait time of " + waitTime + " is not between 0 and 20");
}
try {
List<Message> messageList = queue.receive(maxNumberOfMessages,visibilityTimeout,waitTime);
return new ReceiveMessageResult().withMessages(messageList);
} catch (IOException e) {
throw new AmazonServiceException("error reading messages from " + queue.getQueuePath().toUri().toString(),e);
}
}
项目：NBS3Sync
文件：TestSQSUtils.java
@Test
public void testConvertMessageToEvent() {
Message message = new Message();
message.setBody("{\n" +
" \"Type\" : \"Notification\",\n" +
" \"MessageId\" : \"942c624a-5f4f-5fe4-a2d7-28e60ff48e15\",\n" +
" \"TopicArn\" : \"t\",\n" +
" \"Subject\" : \"Operation\",\n" +
" \"Message\" : \"{\\\"operation\\\":\\\"CREATE\\\",\\\"file\\\":\\\"f4\\\",\\\"originator\\\":\\\"me\\\"}\",\n" +
" \"Timestamp\" : \"2013-09-12T23:30:03.434Z\",\n" +
" \"SignatureVersion\" : \"1\",\n" +
" \"Signature\" : \"n4tdSNZBDlaOgCS2ILxbhCGq3Cl/5LlG2SAh1OKUbat3qWzgtlD5PvgjJUq/heCb5Eo0KnnDYAJbeuiw7X9HmotCtq50OiqCtz6uq6EuApy1LhiBzOhyC5S4yarmQqGV0PBZCGLrvPn/So1HiVsFoBujJZNQuw0ysQQ/ILi6TFA=\",\n" +
" \"SigningCertURL\" : \"cert\",\n" +
" \"UnsubscribeURL\" : \"foo\"\n" +
"}");
message.setReceiptHandle("receiptHandle");
message.setMessageId("messageId");
Event event = SQSUtils.convertMessageToEvent(baseDir,message);
assertEquals("operation is incorrect",FileOperation.CREATE,event.getOperation());
assertEquals("originator is correct","me",event.getOriginator());
assertEquals("file path is correct","/foo/bar/f4",event.getFile().toPath().toString());
assertEquals("isDirectory is correct",false,event.isDirectory());
}
项目：NBS3Sync
文件：TestSQSUtils.java
@Test
public void testConvertMessageToEventIsDirPositive() {
Message message = new Message();
message.setBody("{\n" +
" \"Type\" : \"Notification\",\\\"file\\\":\\\"dir/\\\",\n" +
" \"Signature\" : \"sig/sig=\",\n" +
" \"UnsubscribeURL\" : \"foo\"\n" +
"}");
message.setReceiptHandle("receiptHandle");
message.setMessageId("messageId");
Event event = SQSUtils.convertMessageToEvent(baseDir,"/foo/bar/dir",true,event.isDirectory());
}
项目：s3mper
文件：AlertJanitor.java
/**
* Writes out logs to the given path as a separate JSON message per line.
*
* @param queue
* @param path
* @throws IOException
*/
public void writeLogs(String queue,Path path) throws IOException {
FileSystem fs = FileSystem.get(path.toUri(),conf);
DataOutputStream fout = fs.create(path);
do {
List<Message> messages = pull(queue,batchCount);
if(messages.isEmpty()) {
break;
}
for(Message m : messages) {
fout.write((m.getBody().replaceAll("[\n|\r]"," ")+"\n").getBytes("UTF8"));
}
delete(queue,messages);
} while(true);
fout.close();
fs.close();
}