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

/**
 * Retrieves queue url for the given queue name. If the queue does not exist, tries to create it.
 *
 * @param queueName the queue name to get url for
 * @return an optional String representing the queue url
 */
Optional<String> getUrlForQueue(String queueName) {
    Optional<String> queueUrl = Optional.empty();
    try {
        GetQueueUrlResult queueUrlResult = sqs.getQueueUrl(queueName);
        if (queueUrlResult.getQueueUrl() != null) {
            queueUrl = Optional.of(queueUrlResult.getQueueUrl());
        }
    } catch (QueueDoesNotExistException e) {
        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("Queue " + queueName + " does not exist, try to create it",e);
        }
        CreateQueueRequest createQueueRequest = new CreateQueueRequest(queueName);
        try {
            queueUrl = Optional.of(sqs.createQueue(createQueueRequest).getQueueUrl());
        } catch (AmazonClientException e2) {
            LOGGER.error("Could not create queue " + queueName + ", bundle won't work",e2);
        }
    }

    return queueUrl;
}

@Test
public void shouldCreateNewQueueWhenNoQueueUrlIsFound() throws Exception {
    //GIVEN
    AmazonSQS sqs = mock(AmazonSQS.class);
    field("sqs").ofType(AmazonSQS.class).in(bundle).set(sqs);

    String queueUrl = "https://eu-central-1/queue.amazonaws.com/123456/test-queue";
    when(sqs.getQueueUrl("test-queue")).thenThrow(new QueueDoesNotExistException("Simulates that queue does not exist"));
    when(sqs.createQueue(new CreateQueueRequest("test-queue"))).thenReturn(new CreateQueueResult().withQueueUrl(queueUrl));

    //WHEN
    Optional<String> urlForQueue = bundle.getUrlForQueue("test-queue");

    //THEN
    assertThat(urlForQueue.isPresent()).isTrue();
    assertThat(urlForQueue.get()).isEqualTo(queueUrl);
}

private void initQueue() {
    this.sqs = new AmazonSQSClient(); // Do we need to use new
                                        // ClientConfiguration().withMaxConnections(256)
                                        // ?
    this.sqs.configureRegion(region);
    try {
        // Check to see if queue exists
        GetQueueUrlResult queueUrlResult = this.sqs.getQueueUrl(getSqsQueueName());
        this.queueUrl = queueUrlResult.getQueueUrl();
    } catch (QueueDoesNotExistException queueDoesNotExist) {
        // Queue does not exist, need to create one
        CreateQueueRequest createQueueRequest = new CreateQueueRequest();
        createQueueRequest.setQueueName(getSqsQueueName());
        createQueueRequest.addAttributesEntry("VisibilityTimeout", "" + getVisibilityTimeout());
        CreateQueueResult createQueueResult = this.sqs.createQueue(createQueueRequest);
        this.queueUrl = createQueueResult.getQueueUrl();
    }
}

/**
 * Get a queue url from a queue name
 * @param queueName
 * @return queueUrl - For the specified queue name
 */
private synchronized String getAndSetQueueUrl(final String queueName) throws QueueDoesNotExistException{
    try{

        final String url = queueUrlMap.get(queueName); 
        if(url != null){
            return url;
        }else{
            final GetQueueUrlResult result = this.sqs.getQueueUrl(queueName);
            if(result != null && !Strings.isNullOrEmpty(result.getQueueUrl())){
                    queueUrlMap.put(queueName, result.getQueueUrl());   
                    return result.getQueueUrl();
            }               
        }
    }catch(QueueDoesNotExistException qne){
        throw qne;
    }catch(Exception ex){
        throw new RuntimeException(ex.getMessage(), ex);
    }
    return null;
}

private void createQueueAndConfigIfNotExists(final String queue, final MessageHandler handler) {

    /** determine if the queue exists, if not, create it with the default settings **/
    try{
        final String url = this.admin.getQueueUrl(queue);
        if(Strings.isNullOrEmpty(url)){
            throw new QueueDoesNotExistException(String.format("The queue: %s was not found", queue));
        }
    }catch(QueueDoesNotExistException qneException){
        /** determine if the queue configuration exists **/
        SqsQueueConfig queueConfig = this.queueConfigMap.get(queue);
        if(queueConfig==null){
            /** create default config if we don't know about it **/
            queueConfig = new SqsQueueConfig();
            queueConfig.setName(queue);
            this.queueConfigMap.put(queue, queueConfig);
        }
        LOGGER.info("Queue: {} does not exist - creating the queue now", queue);
        this.admin.initializeQueue(queueConfig, this.config.isBlockUntilReady());
    }
}

@Override
public String resolveDestination(String name) throws DestinationResolutionException {
    String queueName = name;

    if (this.resourceIdResolver != null) {
        queueName = this.resourceIdResolver.resolveToPhysicalResourceId(name);
    }

    if (isValidQueueUrl(queueName)) {
        return queueName;
    }

    if (this.autoCreate) {
        //Auto-create is fine to be called even if the queue exists.
        CreateQueueResult createQueueResult = this.amazonSqs.createQueue(new CreateQueueRequest(queueName));
        return createQueueResult.getQueueUrl();
    } else {
        try {
            GetQueueUrlResult getQueueUrlResult = this.amazonSqs.getQueueUrl(new GetQueueUrlRequest(queueName));
            return getQueueUrlResult.getQueueUrl();
        } catch (QueueDoesNotExistException e) {
            throw new DestinationResolutionException(e.getMessage(), e);
        }
    }
}

private String queryQueueUrl(String queueName) {
    try {
        return _sqs.getQueueUrl(new GetQueueUrlRequest(queueName)).getQueueUrl();
    } catch (QueueDoesNotExistException e) {
        // Create the queue
        int visibilityTimeout = queueName.equals(_pendingScanRangeQueue) ?
                DEFAULT_TASK_CLAIM_VISIBILITY_TIMEOUT : DEFAULT_TASK_COMPLETE_VISIBILITY_TIMEOUT;
        return _sqs.createQueue(
                new CreateQueueRequest(queueName)
                        .withAttributes(ImmutableMap.<String, String>of(
                                "VisibilityTimeout", String.valueOf(visibilityTimeout)))
        ).getQueueUrl();
    }
}

@Override
protected int poll() throws Exception {
    // must reset for each poll
    shutdownRunningTask = null;
    pendingExchanges = 0;

    ReceiveMessageRequest request = new ReceiveMessageRequest(getQueueUrl());
    request.setMaxNumberOfMessages(getMaxMessagesPerPoll() > 0 ? getMaxMessagesPerPoll() : null);
    request.setVisibilityTimeout(getConfiguration().getVisibilityTimeout() != null ? getConfiguration().getVisibilityTimeout() : null);
    request.setWaitTimeSeconds(getConfiguration().getWaitTimeSeconds() != null ? getConfiguration().getWaitTimeSeconds() : null);

    if (attributeNames != null) {
        request.setAttributeNames(attributeNames);
    }
    if (messageAttributeNames != null) {
        request.setMessageAttributeNames(messageAttributeNames);
    }

    LOG.trace("Receiving messages with request [{}]...", request);

    ReceiveMessageResult messageResult = null;
    try {
        messageResult = getClient().receiveMessage(request);
    } catch (QueueDoesNotExistException e) {
        LOG.info("Queue does not exist....recreating now...");
        reConnectToQueue();
        messageResult = getClient().receiveMessage(request);
    }

    if (LOG.isTraceEnabled()) {
        LOG.trace("Received {} messages", messageResult.getMessages().size());
    }

    Queue<Exchange> exchanges = createExchanges(messageResult.getMessages());
    return processBatch(CastUtils.cast(exchanges));
}

private String getQueueUrl(String queueName, boolean createIfNotExist) {
    try {
        return _sqs.getQueueUrl(queueName).getQueueUrl();
    } catch (QueueDoesNotExistException e) {
        if (createIfNotExist) {
            logger.info("Creating SQS queue called: " + queueName);
            return createQueue(queueName);
        }

        throw e;
    }
}

public boolean stillExists() {
    try {
        _sqs.getQueueUrl(_queueName);
        return true;
    } catch (QueueDoesNotExistException e) {
        return false;
    }
}

@Override
public SendMessageResult sendMessage(String queueName, String messageText, Map<String, MessageAttributeValue> messageAttributes, AmazonSQS amazonSQS)
{
    try
    {
        return amazonSQS.sendMessage(new SendMessageRequest().withQueueUrl(amazonSQS.getQueueUrl(queueName).getQueueUrl()).withMessageBody(messageText)
            .withMessageAttributes(messageAttributes));
    }
    catch (QueueDoesNotExistException e)
    {
        throw new IllegalStateException(String.format("AWS SQS queue with \"%s\" name not found.", queueName), e);
    }
}

@Test(expected = CannotCreateSenderException.class)
public void shouldThrowExceptionWhenCreatingSenderIfQueueDoesNotExists() throws Exception, CannotCreateSenderException {
    //GIVEN
    AmazonSQS sqs = mock(AmazonSQS.class);
    field("sqs").ofType(AmazonSQS.class).in(bundle).set(sqs);

    when(sqs.getQueueUrl(anyString())).thenThrow(new QueueDoesNotExistException("Simulate queue does not exist"));
    when(sqs.createQueue((CreateQueueRequest) any())).thenThrow(new AmazonClientException("Simulate queue cannot be created"));

    //WHEN
    bundle.createSender("test-queue");

    //THEN
}

@PostConstruct
public void postConstruct() {
    // for each configured queue set up the data structure to manage the current message batch

    String queuesProperty = StringUtils.trim(config.getString(QUEUE_NAMES_PROPERTY));
    Validate.notEmpty(queuesProperty);

    String[] queues = StringUtils.split(queuesProperty, "| ");

    // Initialize the message and queue URLs
    Map<String, LinkedBlockingQueue<Message>> tempMessagesMap = new HashMap<>(queues.length);
    Map<String, String> tempQueueUrls = new HashMap<>(queues.length);

    for (String queue : queues) {
        queue = StringUtils.trim(queue);
        String queueUrl;

        logger.info("Initializing queue " + queue);

        try {
            queueUrl = sqsClient.getQueueUrl(queue).getQueueUrl();
        } catch (QueueDoesNotExistException ex) {
            queueUrl = sqsClient.createQueue(queue).getQueueUrl();
        }

        tempMessagesMap.put(queue, new LinkedBlockingQueue<>());
        tempQueueUrls.put(queue, queueUrl);
    }

    messagesMap = Collections.unmodifiableMap(tempMessagesMap);
    queueUrls = Collections.unmodifiableMap(tempQueueUrls);

    // Start the async operation
    executorService.submit(messagePoller);
}

/**
 * Get the queue url. First an internal cache is checked, if the name to queueUrl mapping
 * is not found in the internal cache a call is made to the Sqs API. If a valid queue
 * url is returned the name -> queueUrl mapping will be cached locally 
 * @param queueName
 * @return queueUrl
 */
public String getQueueUrl(final String queueName) throws QueueDoesNotExistException{

    Preconditions.checkArgument(!Strings.isNullOrEmpty(queueName), "Queue Name is required in order to submit a message for sqs");
    String url = queueUrlMap.get(queueName);
    if(url!=null){
        return url;
    }else{
        return this.getAndSetQueueUrl(queueName);
    }
}

@Test
public void testInvalidDestinationName() throws Exception {
    AmazonSQS amazonSqs = mock(AmazonSQS.class);
    AmazonServiceException exception = new QueueDoesNotExistException("AWS.SimpleQueueService.NonExistentQueue");
    exception.setErrorCode("AWS.SimpleQueueService.NonExistentQueue");
    String queueUrl = "invalidName";
    when(amazonSqs.getQueueUrl(new GetQueueUrlRequest(queueUrl))).thenThrow(exception);
    DynamicQueueUrlDestinationResolver dynamicQueueDestinationResolver = new DynamicQueueUrlDestinationResolver(amazonSqs);
    try {
        dynamicQueueDestinationResolver.resolveDestination(queueUrl);
    } catch (DestinationResolutionException e) {
        assertTrue(e.getMessage().startsWith("AWS.SimpleQueueService.NonExistentQueue"));
    }
}

@Test(expected = InvalidDestinationException.class)
public void testGetQueueUrlQueueNameThrowQueueDoesNotExistException() throws JMSException {

    GetQueueUrlRequest getQueueUrlRequest = new GetQueueUrlRequest(QUEUE_NAME);
    doThrow(new QueueDoesNotExistException("qdnee"))
            .when(amazonSQSClient).getQueueUrl(eq(getQueueUrlRequest));

    wrapper.getQueueUrl(QUEUE_NAME);
}

@Test(expected = InvalidDestinationException.class)
public void testGetQueueUrlQueueNameWithAccountIdThrowQueueDoesNotExistException() throws JMSException {

    GetQueueUrlRequest getQueueUrlRequest = new GetQueueUrlRequest(QUEUE_NAME);
    getQueueUrlRequest.setQueueOwnerAWSAccountId(OWNER_ACCOUNT_ID);
    doThrow(new QueueDoesNotExistException("qdnee"))
            .when(amazonSQSClient).getQueueUrl(eq(getQueueUrlRequest));

    wrapper.getQueueUrl(QUEUE_NAME,OWNER_ACCOUNT_ID);
}

@Test
public void testQueueExistsThrowQueueDoesNotExistException() throws JMSException {

    GetQueueUrlRequest getQueueUrlRequest = new GetQueueUrlRequest(QUEUE_NAME);
    doThrow(new QueueDoesNotExistException("qdnee"))
            .when(amazonSQSClient).getQueueUrl(eq(getQueueUrlRequest));

    assertFalse(wrapper.queueExists(QUEUE_NAME));
}

@Override
public GetQueueUrlResult getQueueUrl(GetQueueUrlRequest getQueueUrlRequest) throws AmazonClientException {
    try {
        File topicFile = new File(_rootDirectory, getQueueUrlRequest.getQueueName());
        if (!topicFile.exists()) {
            throw new QueueDoesNotExistException("could not find a file for queue named " + getQueueUrlRequest.getQueueName());
        }
        return new GetQueueUrlResult().withQueueUrl(saveQueue(new DirectorySQSQueue(topicFile.toPath())));
    } catch (IOException e) {
        throw new AmazonServiceException("could not get queue named " + getQueueUrlRequest.getQueueName(), e);
    }
}

@Test(expectedExceptions = QueueDoesNotExistException.class)
public void cannotDeleteNonExistentQueue()
        throws IOException {
    _amazonSQS.deleteQueue(new DeleteQueueRequest(new File(TestUtils.createTempDirectory(), someQueueName()).toURI().toString()));
}

@Test(expectedExceptions = QueueDoesNotExistException.class)
public void failsOnNonExistentQueue() {
    _amazonSQS.getQueueUrl(new GetQueueUrlRequest(someQueueName()));
}