Java 类com.amazonaws.services.sqs.model.AmazonSQSException 实例源码

/**
 * 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);
    });
}

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;
}

@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
}

@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));
}

@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
}

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);
}

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());
    }
}

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);
}