项目:async-sqs
文件:SqsClient.java
/**
* Asserts that an SQS queue exists with specific attributes. The queue is created if it does not exist,* and any existing queue is modified if the attributes don't match.
*
* @param queueConfig Configuration of the SQS queue
* @param clientConfig Configuration of the SQS queue client
* @return an SqsQueue
*/
public Single<SqsQueue<String>> upsertQueue(SqsQueueConfig queueConfig,SqsQueueClientConfig clientConfig) {
CreateQueueAction action = new CreateQueueAction(queueConfig);
Single<SqsQueue<String>> output = requestSender.sendRequest(action).map(createQueueResult -> {
return new BufferedStringSqsQueue(createQueueResult.getQueueUrl(),requestSender,clientConfig);
});
return output.onErrorResumeNext((err) -> {
if (err instanceof AmazonSQSException) {
AmazonSQSException awsException = (AmazonSQSException) err;
//Queue already exists,but has wrong attributes. We need to update them.
if (QUEUE_ALREADY_EXISTS.equals(awsException.getErrorCode())) {
//Have to get queue from name since we don't know the url yet.
return getQueueFromName(queueConfig.getName(),queueConfig.getRegion(),clientConfig)
.flatMap((queue) -> {
return queue.setAttributes(queueConfig.getAttributes()).toSingleDefault(queue);
});
}
}
return Single.error(err);
});
}
项目:async-sqs
文件:RetryingSqsQueue.java
private boolean shouldRetry(int errCount,Throwable error) {
if (errCount > retryCount) {
return false;
}
if (error instanceof AmazonSQSException) {
return ((AmazonSQSException) error).getErrorType() == AmazonServiceException.ErrorType.Service;
}
return true;
}
项目:async-sqs
文件:RetryingSqsRequestSender.java
@Override
public <T> Single<T> sendRequest(SqsAction<T> request) {
return Single.defer(() -> delegate.sendRequest(request))
.retry((errCount,error) -> {
if (errCount > retryCount || request.isBatchAction()) {
return false;
}
if (error instanceof AmazonSQSException) {
return ((AmazonSQSException) error).getErrorType() == AmazonServiceException.ErrorType.Service;
}
return true;
}).subscribeWith(SingleSubject.create());//convert to Hot single
}
项目:async-sqs
文件:SqsClientTest.java
@Test
public void testAssertQueueUnknownAmazonError() {
when(requestSenderMock.sendRequest(any(CreateQueueAction.class))).thenReturn(Single.error(
new AmazonSQSException("Unknown error")
));
client.upsertQueue(QUEUE_CONFIG).test().assertError(RuntimeException.class);
verify(requestSenderMock).sendRequest(any(CreateQueueAction.class));
}
项目:async-sqs
文件:RetryingSqsRequestSenderTest.java
@Test
public void testRetryForServerError() {
when(delegateMock.sendRequest(any()))
.thenThrow(SERVER_EXCEPTION);
requestSender.sendRequest(actionMock).test().assertError(AmazonSQSException.class);
verify(delegateMock,times(2)).sendRequest(any());//exactly 2 requests sent
}
项目:aws-doc-sdk-examples
文件:VisibilityTimeout.java
public static void main(String[] args)
{
final String queue_name = "testQueue" + new Date().getTime();
AmazonSQS sqs = AmazonSQSClientBuilder.defaultClient();
// first,create a queue (unless it exists already)
try {
CreateQueueResult cq_result = sqs.createQueue(queue_name);
} catch (AmazonSQSException e) {
if (!e.getErrorCode().equals("QueueAlreadyExists")) {
throw e;
}
}
final String queue_url = sqs.getQueueUrl(queue_name).getQueueUrl();
// Send some messages to the queue
for (int i = 0; i < 20; i++) {
sqs.sendMessage(queue_url,"This is message " + i);
}
// change visibility timeout (single)
changeMessageVisibilitySingle(queue_url,3600);
// change visibility timeout (multiple)
changeMessageVisibilityMultiple(queue_url,2000);
}
项目:aws-doc-sdk-examples
文件:SendReceiveMessages.java
public static void main(String[] args)
{
final AmazonSQS sqs = AmazonSQSClientBuilder.defaultClient();
try {
CreateQueueResult create_result = sqs.createQueue(QUEUE_NAME);
} catch (AmazonSQSException e) {
if (!e.getErrorCode().equals("QueueAlreadyExists")) {
throw e;
}
}
String queueUrl = sqs.getQueueUrl(QUEUE_NAME).getQueueUrl();
SendMessageRequest send_msg_request = new SendMessageRequest()
.withQueueUrl(queueUrl)
.withMessageBody("hello world")
.withDelaySeconds(5);
sqs.sendMessage(send_msg_request);
// Send multiple messages to the queue
SendMessageBatchRequest send_batch_request = new SendMessageBatchRequest()
.withQueueUrl(queueUrl)
.withEntries(
new SendMessageBatchRequestEntry(
"msg_1","Hello from message 1"),new SendMessageBatchRequestEntry(
"msg_2","Hello from message 2")
.withDelaySeconds(10));
sqs.sendMessageBatch(send_batch_request);
// receive messages from the queue
List<Message> messages = sqs.receiveMessage(queueUrl).getMessages();
// delete messages from the queue
for (Message m : messages) {
sqs.deleteMessage(queueUrl,m.getReceiptHandle());
}
}
项目:aws-doc-sdk-examples
文件:LongPolling.java
public static void main(String[] args)
{
final String USAGE =
"To run this example,supply the name of a queue to create and\n" +
"queue url of an existing queue.\n\n" +
"Ex: LongPolling <unique-queue-name> <existing-queue-url>\n";
if (args.length != 2) {
System.out.println(USAGE);
System.exit(1);
}
String queue_name = args[0];
String queue_url = args[1];
final AmazonSQS sqs = AmazonSQSClientBuilder.defaultClient();
// Enable long polling when creating a queue
CreateQueueRequest create_request = new CreateQueueRequest()
.withQueueName(queue_name)
.addAttributesEntry("ReceiveMessageWaitTimeSeconds","20");
try {
sqs.createQueue(create_request);
} catch (AmazonSQSException e) {
if (!e.getErrorCode().equals("QueueAlreadyExists")) {
throw e;
}
}
// Enable long polling on an existing queue
SetQueueAttributesRequest set_attrs_request = new SetQueueAttributesRequest()
.withQueueUrl(queue_url)
.addAttributesEntry("ReceiveMessageWaitTimeSeconds","20");
sqs.setQueueAttributes(set_attrs_request);
// Enable long polling on a message receipt
ReceiveMessageRequest receive_request = new ReceiveMessageRequest()
.withQueueUrl(queue_url)
.withWaitTimeSeconds(20);
sqs.receiveMessage(receive_request);
}