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

@Test
public void testPurgeQueue_shouldRemoveAll() {
  // create queue
  CreateQueueResult createdQueue = sqs.createQueue(new CreateQueueRequest().withQueueName("tea-earl-grey-queue"));
  // send messages
  String messageBody = "{\"life-universe-everything\":42}";
  sqs.sendMessage(new SendMessageRequest().withDelaySeconds(0).withMessageBody(messageBody)
      .withMessageGroupId("some-group-id-123").withQueueUrl(createdQueue.getQueueUrl()));
  String messageBody2 = "{\"dead-emptyness-nothing\":24}";
  sqs.sendMessage(new SendMessageRequest().withDelaySeconds(0).withMessageBody(messageBody2)
      .withMessageGroupId("some-group-id-123").withQueueUrl(createdQueue.getQueueUrl()));

  // purge queues
  PurgeQueueResult result = sqs.purgeQueue(new PurgeQueueRequest().withQueueUrl(createdQueue.getQueueUrl()));
  assertNotNull("verify that purge queue returned ok", result);

  // verify empty queue
  ReceiveMessageResult messageResult = sqs.receiveMessage(new ReceiveMessageRequest()
      .withMaxNumberOfMessages(9).withQueueUrl(createdQueue.getQueueUrl()).withVisibilityTimeout(10)
      .withWaitTimeSeconds(0));
  assertEquals("verify that queue is empty", 0, messageResult.getMessages().size());

  // cleanup
  getQueues().remove("tea-earl-grey-queue");
}

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

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

@Override
public void initialize(JSONObject config) throws Exception {
    String endpoint = config.getString(PARAM_ENDPOINT);
    String queueName = config.getString(GenericQueue.PARAM_NAME);
    String accessKey = config.getString(PARAM_ACCESS_KEY);
    String secretKey = config.getString(PARAM_SECRET_KEY);
    if(accessKey==null) throw new Exception(PARAM_ACCESS_KEY+" is required!");
    if(secretKey==null) throw new Exception(PARAM_SECRET_KEY+" is required!");
    try {
        this.client = new AmazonSQSAsyncClient(new BasicAWSCredentials(accessKey,secretKey));
        this.client.setEndpoint(endpoint);
        CreateQueueRequest createQueueRequest = new CreateQueueRequest(queueName);
        CreateQueueResult createQueueResult = this.client.createQueue(createQueueRequest);
        this.queueUrl = createQueueResult.getQueueUrl();
    } catch (Throwable e){
        throw new Exception(e);
    }
}

/**
 * Creates a new SQS queue on AWS.
 * @param name queue name
 * @return the queue URL or null
 */
public static String createQueue(String name) {
    if (StringUtils.isBlank(name)) {
        return null;
    }
    String queueURL = getQueueURL(name);
    if (queueURL == null) {
        try {
            queueURL = getClient().createQueue(new CreateQueueRequest(name)).getQueueUrl();
        } catch (AmazonServiceException ase) {
            logException(ase);
        } catch (AmazonClientException ace) {
            logger.error("Could not reach SQS. {0}", ace.toString());
        }
    }
    return queueURL;
}

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

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

public static int getCount(String name) {
    try {
        String queueUrl = getConnection().createQueue(
                new CreateQueueRequest(name)).getQueueUrl();


        List<String> attributeNames = new ArrayList<String>();
        attributeNames.add("All");

        // list the attributes of the queue we are interested in
        GetQueueAttributesRequest request = new GetQueueAttributesRequest(queueUrl);
        request.setAttributeNames(attributeNames);
        Map<String, String> attributes = sqs.getQueueAttributes(request).getAttributes();
        int messages = Integer.parseInt(attributes.get("ApproximateNumberOfMessages"));
        //System.out.println("Messages in the queue: " + messages);
        return messages;

    } catch (Exception e) {
        e.printStackTrace();
        return -1;
    }
}

@VisibleForTesting
static CreateQueueRequest createRequest(SqsQueueConfig config) {
    //TODO: add deadletter config

    return new CreateQueueRequest()
            .withQueueName(config.getName())
            .withAttributes(config.getAttributes().getStringMap());
}

@Test
public void testCreateGetUrlListQueue_shouldCreateReturnUrlAndListQueue() {
  // create first queue
  CreateQueueResult createdQueue = sqs.createQueue(new CreateQueueRequest().withQueueName("tea-earl-grey-queue"));
  assertNotNull("verify that, on creation, queue url was returned",createdQueue.getQueueUrl());
  // create other queues
  CreateQueueResult secondTeaQueue = sqs.createQueue(new CreateQueueRequest().withQueueName("tea-mate-queue"));
  CreateQueueResult anotherQueue = sqs.createQueue(new CreateQueueRequest().withQueueName("coffee-queue"));
  // get queue url
  GetQueueUrlResult queueUrlResult = sqs.getQueueUrl(new GetQueueUrlRequest()
      .withQueueName("tea-earl-grey-queue").withQueueOwnerAWSAccountId("some owner"));
  assertNotNull("verify that, on fetch, queue url was returned", queueUrlResult.getQueueUrl());
  // get all queues
  ListQueuesResult allQueues = sqs.listQueues();
  assertEquals("verify all queues are returned", 3, allQueues.getQueueUrls().size());
  assertTrue("verify that all queues contain first queue", allQueues.getQueueUrls().contains(createdQueue.getQueueUrl()));
  assertTrue("verify that all queues contain second tea queue", allQueues.getQueueUrls().contains(secondTeaQueue.getQueueUrl()));
  assertTrue("verify that all queues contain coffee queue", allQueues.getQueueUrls().contains(anotherQueue.getQueueUrl()));
  // get only queues that start with 'tea'
  ListQueuesResult teaQueues = sqs.listQueues(new ListQueuesRequest("tea"));
  assertEquals("verify only tea queues are returned", 2, teaQueues.getQueueUrls().size());
  assertTrue("verify that tea queues contain first queue", teaQueues.getQueueUrls().contains(createdQueue.getQueueUrl()));
  assertTrue("verify that tea queues contain second tea queue", teaQueues.getQueueUrls().contains(secondTeaQueue.getQueueUrl()));


  assertNotNull("verify that delete queue returned ok", sqs.deleteQueue(new DeleteQueueRequest().withQueueUrl(queueUrlResult.getQueueUrl())));
  assertFalse("verify that the queue was removed", sqs.listQueues().getQueueUrls().stream()
      .anyMatch( queueUrl -> StringUtils.equals(queueUrl,queueUrlResult.getQueueUrl()) ));

  // cleanup
  getQueues().remove("tea-earl-grey-queue");
  getQueues().remove("tea-mate-queue");
  getQueues().remove("coffee-queue");
}

@VisibleForTesting
String getOrCreateQueue() {
       List<String> queueUrls = listQueues(queueName);
    if (queueUrls == null || queueUrls.isEmpty()) {         
           CreateQueueRequest createQueueRequest = new CreateQueueRequest().withQueueName(queueName);
           CreateQueueResult result = client.createQueue(createQueueRequest);
           return result.getQueueUrl();
    } else {
           return queueUrls.get(0);
       }
   }

SendMessageResult createQueueMessage(ImageCreateRequest request, Context context) {

        LambdaLogger logger = context.getLogger();

        final String queueName = request.getQueueName();
        final String sqsEndpoint = request.getSqsEndpoint();

        if (queueName == null || sqsEndpoint == null) {
            logger.log("skip create queue. [" + request + "]");
            return null;
        }

        AmazonSQSAsync client = createSQSClient();
        client.setEndpoint(sqsEndpoint);
        request.setSendMessageTimeMillis(System.currentTimeMillis());

        try {
            CreateQueueRequest req = new CreateQueueRequest(queueName);
            String queueUrl = client.createQueue(req).getQueueUrl();

            SendMessageRequest sendMessage = new SendMessageRequest();
            sendMessage.setQueueUrl(queueUrl);
            ObjectMapper om = new ObjectMapper();
            sendMessage.setMessageBody(om.writeValueAsString(request));

            Future<SendMessageResult> result = client.sendMessageAsync(sendMessage);
            while (!result.isDone()) {
                Thread.sleep(100);
            }
            return result.get();

        } catch (Exception e) {
            throw new RuntimeException("unexpected error occured in the create queue request.", e);
        } finally {
            client.shutdown();
        }
    }

SendMessageResult createQueueMessage(InstanceRequest instanceRequest, Context context) {

        LambdaLogger logger = context.getLogger();

        final String queueName = instanceRequest.getQueueName();
        final String sqsEndpoint = instanceRequest.getSqsEndpoint();

        if (queueName == null || sqsEndpoint == null) {
            logger.log("skip create queue. instanceRequest[" + instanceRequest + "]");
            return null;
        }

        AmazonSQSAsync client = createSQSClient();
        client.setEndpoint(sqsEndpoint);

        try {
            CreateQueueRequest req = new CreateQueueRequest(queueName);
            String queueUrl = client.createQueue(req).getQueueUrl();

            instanceRequest.setSendMessageTimeMillis(System.currentTimeMillis());

            SendMessageRequest sendMessage = new SendMessageRequest();
            sendMessage.setQueueUrl(queueUrl);
            ObjectMapper om = new ObjectMapper();
            sendMessage.setMessageBody(om.writeValueAsString(instanceRequest));

            Future<SendMessageResult> result = client.sendMessageAsync(sendMessage);
            while (!result.isDone()) {
                Thread.sleep(100);
            }
            return result.get();

        } catch (Exception e) {
            throw new RuntimeException("unexpected error occured in the create queue request.", e);
        } finally {
            client.shutdown();
        }
    }

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

protected void createQueue(AmazonSQS client) {
    LOG.trace("Queue '{}' doesn't exist. Will create it...", configuration.getQueueName());

    // creates a new queue, or returns the URL of an existing one
    CreateQueueRequest request = new CreateQueueRequest(configuration.getQueueName());
    if (getConfiguration().getDefaultVisibilityTimeout() != null) {
        request.getAttributes().put(QueueAttributeName.VisibilityTimeout.name(), String.valueOf(getConfiguration().getDefaultVisibilityTimeout()));
    }
    if (getConfiguration().getMaximumMessageSize() != null) {
        request.getAttributes().put(QueueAttributeName.MaximumMessageSize.name(), String.valueOf(getConfiguration().getMaximumMessageSize()));
    }
    if (getConfiguration().getMessageRetentionPeriod() != null) {
        request.getAttributes().put(QueueAttributeName.MessageRetentionPeriod.name(), String.valueOf(getConfiguration().getMessageRetentionPeriod()));
    }
    if (getConfiguration().getPolicy() != null) {
        request.getAttributes().put(QueueAttributeName.Policy.name(), String.valueOf(getConfiguration().getPolicy()));
    }
    if (getConfiguration().getReceiveMessageWaitTimeSeconds() != null) {
        request.getAttributes().put(QueueAttributeName.ReceiveMessageWaitTimeSeconds.name(), String.valueOf(getConfiguration().getReceiveMessageWaitTimeSeconds()));
    }
    if (getConfiguration().getRedrivePolicy() != null) {
        request.getAttributes().put(QueueAttributeName.RedrivePolicy.name(), getConfiguration().getRedrivePolicy());
    }
    LOG.trace("Creating queue [{}] with request [{}]...", configuration.getQueueName(), request);

    CreateQueueResult queueResult = client.createQueue(request);
    queueUrl = queueResult.getQueueUrl();

    LOG.trace("Queue created and available at: {}", queueUrl);
}

@Override
public CreateQueueResult createQueue(CreateQueueRequest createQueueRequest) throws AmazonServiceException, AmazonClientException {
    String queueName = "https://queue.amazonaws.com/541925086079/" + createQueueRequest.getQueueName();
    queues.put(queueName, createQueueRequest);
    CreateQueueResult result = new CreateQueueResult();
    result.setQueueUrl(queueName);
    return result;
}

/**
 * create a queue we can use for testing
 *
 * @throws Exception
 */
@Before
public void beforTest() throws Exception
{
  validateAssumption();
  // Create a queue
  CreateQueueRequest createQueueRequest = new CreateQueueRequest().withQueueName(getCurrentQueueName());
  currentQueueUrl = sqs.createQueue(createQueueRequest).getQueueUrl();
}

private String createQueue (String queueName) {
    logger.info("Getting SQS queue called: " + queueName);
       CreateQueueRequest createQueueRequest = new CreateQueueRequest(queueName);
    Map<String,String> m = new HashMap<String,String>();
    m.put("VisibilityTimeout", "60");
       createQueueRequest.setAttributes(m);
       return _sqs.createQueue(createQueueRequest).getQueueUrl();
}

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

@Test
public void createQueue() {
  final String queueName = "bizo";

  final CreateQueueRequest createQueueRequest = new CreateQueueRequest().withQueueName(queueName);
  sqs.createQueue(createQueueRequest);

  final ListQueuesResult listQueuesResult = sqs.listQueues();
  assertThat(listQueuesResult.getQueueUrls(), hasItem(containsString(queueName)));
}

@Test
public void getQueueUrl() {
  final String queueName = "bizo";

  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();
  assertThat(queueUrl, containsString(queueName));
}

@Test
public void deleteMessageSucceedsWithValidReceiptHandle() {
  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));

  final String receiptHandle = messages.get(0).getReceiptHandle();
  final DeleteMessageRequest deleteMessageRequest =
    new DeleteMessageRequest().withQueueUrl(queueUrl).withReceiptHandle(receiptHandle);

  try {
    sqs.deleteMessage(deleteMessageRequest);
  } catch (ReceiptHandleIsInvalidException e) {
    fail("ReceiptHandleIsInvalidException was thrown");
  }
}

@Test(expected = ReceiptHandleIsInvalidException.class)
public void deleteMessageFailsWithInvalidReceiptHandle() {
  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 ReceiveMessageRequest receiveMessageRequest = new ReceiveMessageRequest().withQueueUrl(queueUrl);
  final ReceiveMessageResult receiveMessageResult = sqs.receiveMessage(receiveMessageRequest);
  final List<Message> messages = receiveMessageResult.getMessages();
  assertThat(messages.size(), equalTo(1));

  final String receiptHandle = "bizo";
  final DeleteMessageRequest deleteMessageRequest =
    new DeleteMessageRequest().withQueueUrl(queueUrl).withReceiptHandle(receiptHandle);

  sqs.deleteMessage(deleteMessageRequest);
}

public SQSService(String queueName){
    /*
        * The ProfileCredentialsProvider will return your [default]
        * credential profile by reading from the credentials file located at
        * (~/.aws/credentials).
        */
       AWSCredentials credentials = null;
       try {
           credentials = new ProfileCredentialsProvider().getCredentials();
       } catch (Exception e) {
           throw new AmazonClientException(
                   "Cannot load the credentials from the credential profiles file. " +
                   "Please make sure that your credentials file is at the correct " +
                   "location (~/.aws/credentials), and is in valid format.",
                   e);
       }

       sqs = new AmazonSQSClient(credentials);
       Region usEast1 = Region.getRegion(Regions.US_EAST_1);
    sqs.setRegion(usEast1);

    // Create a queue or returns the URL of an existing one
       //System.out.println("Creating a new SQS queue called " + queueName);
       CreateQueueRequest createQueueRequest = new CreateQueueRequest(queueName);
       queueUrl = sqs.createQueue(createQueueRequest).getQueueUrl();

}

/**
 * Initialize Queues
 * @param queue
 */
public void initializeQueue(SqsQueueConfig queue, final boolean blockUntilReady){

    final CreateQueueRequest request = new CreateQueueRequest(queue.getName())
    .withAttributes(queue.createAttributes());      
    this.sqs.createQueue(request);

    /** 
     * if blockUntilReady is true, Ensure the queue is created before returning
     * **/
    if(blockUntilReady){
        this.ensureQueueIsReady(queue);
    }               
}

@Override
public Queue createQueue(CreateQueueRequest request,
        ResultCapture<CreateQueueResult> extractor) {

    ActionResult result = service.performAction("CreateQueue", request,
            extractor);

    if (result == null) return null;
    return new QueueImpl(result.getResource());
}

@Override
public Queue createQueue(String queueName, ResultCapture<CreateQueueResult>
        extractor) {

    CreateQueueRequest request = new CreateQueueRequest()
        .withQueueName(queueName);
    return createQueue(request, extractor);
}

@Test
public void testAutoCreate() throws Exception {
    AmazonSQS amazonSqs = mock(AmazonSQS.class);
    String queueUrl = "http://foo/bar";
    when(amazonSqs.createQueue(new CreateQueueRequest("foo"))).thenReturn(new CreateQueueResult().withQueueUrl(queueUrl));

    DynamicQueueUrlDestinationResolver dynamicQueueDestinationResolver = new DynamicQueueUrlDestinationResolver(amazonSqs);
    dynamicQueueDestinationResolver.setAutoCreate(true);
    assertEquals(queueUrl, dynamicQueueDestinationResolver.resolveDestination("foo"));
}

public static boolean pushTo(String name, String message) {
    try {
        String queueUrl = getConnection().createQueue(
                new CreateQueueRequest(name)).getQueueUrl();
        SendMessageRequest sendMessageRequest = new SendMessageRequest(
                queueUrl, message);
        if (null != sendMessageRequest) {
            getConnection().sendMessage(sendMessageRequest);
            return true;
        }
    } catch (Exception e) {
        e.printStackTrace();
    }
    return false;
}

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

    doThrow(new AmazonClientException("ace"))
            .when(amazonSQSClient).createQueue(eq(new CreateQueueRequest(QUEUE_NAME)));

    wrapper.createQueue(QUEUE_NAME);
}

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

    doThrow(new AmazonServiceException("ase"))
            .when(amazonSQSClient).createQueue(eq(new CreateQueueRequest(QUEUE_NAME)));

    wrapper.createQueue(QUEUE_NAME);
}

@Test
public void testCreateQueue() throws JMSException {

    CreateQueueRequest createQueueRequest = new CreateQueueRequest(QUEUE_NAME);

    wrapper.createQueue(createQueueRequest);
    verify(amazonSQSClient).createQueue(createQueueRequest);
}

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

    CreateQueueRequest createQueueRequest = new CreateQueueRequest(QUEUE_NAME);
    doThrow(new AmazonClientException("ace"))
            .when(amazonSQSClient).createQueue(eq(createQueueRequest));

    wrapper.createQueue(createQueueRequest);
}

@Override
public CreateQueueResult createQueue(CreateQueueRequest createQueueRequest) throws AmazonClientException {
    try {
        File topicFile = new File(_rootDirectory, createQueueRequest.getQueueName());
        if (topicFile.exists()) {
            throw new QueueNameExistsException("File exists: " + topicFile);
        }
        Files.createDirectory(topicFile.toPath());
        return new CreateQueueResult().withQueueUrl(saveQueue(new DirectorySQSQueue(topicFile.toPath())));
    } catch (IOException e) {
        throw new AmazonServiceException("could not create a queue named " + createQueueRequest.getQueueName(), e);
    }
}

public void canCreateQueue() {
    final String queueName = someQueueName();

    final CreateQueueResult queue = _amazonSQS.createQueue(new CreateQueueRequest(queueName));
    Assert.assertNotNull(queue.getQueueUrl(), "Queue URL should be present");

    final GetQueueUrlResult result = _amazonSQS.getQueueUrl(new GetQueueUrlRequest(queueName));
    Assert.assertEquals(result.getQueueUrl(), queue.getQueueUrl());
}

@Test(expectedExceptions = QueueNameExistsException.class)
public void cannotRecreateQueue() {
    final String queueName = someQueueName();

    _amazonSQS.createQueue(new CreateQueueRequest(queueName));

    // this should fail
    _amazonSQS.createQueue(new CreateQueueRequest(queueName));
}

public void getQueueArnFromAttributes() {
    String queueName = someQueueName();
    CreateQueueResult createQueueResult = _amazonSQS.createQueue(new CreateQueueRequest(queueName));
    String queueUrl = createQueueResult.getQueueUrl();

    List<String> requestedAttributes = ImmutableList.of("QueueArn");
    GetQueueAttributesResult getQueueAttributesResult = _amazonSQS.getQueueAttributes(new GetQueueAttributesRequest()
            .withQueueUrl(queueUrl)
            .withAttributeNames(requestedAttributes));
    Map<String, String> resultAttributes = getQueueAttributesResult.getAttributes();
    String queueArn = resultAttributes.get("QueueArn");
    String queueNameFromArn = queueArn.substring(queueArn.lastIndexOf(":") + 1);

    Assert.assertEquals(queueNameFromArn, queueName);
}

private void createQueueIfNotExists() {
    for (String qUrl : sqsClient.listQueues().getQueueUrls()) {
        if (qUrl.contains(queueName)) {
            queueUrl = qUrl;
            break;
        }
    }
    if (queueUrl == null) {
        CreateQueueRequest request = new CreateQueueRequest(queueName);
        Map<String, String> queueAttributes = new HashMap<String, String>();
        queueAttributes.put("ReceiveMessageWaitTimeSeconds", Integer
                .valueOf(receiveMessageWaitTimeout).toString());
        if (messageDelay != null) {
            queueAttributes.put("DelaySeconds", messageDelay.toString());
        }
        if (maximumMessageSize != null) {
            queueAttributes.put("MaximumMessageSize",
                    maximumMessageSize.toString());
        }
        if (messageRetentionPeriod != null) {
            queueAttributes.put("MessageRetentionPeriod",
                    messageRetentionPeriod.toString());
        }
        if (visibilityTimeout != null) {
            queueAttributes.put("VisibilityTimeout",
                    visibilityTimeout.toString());
        }
        request.setAttributes(queueAttributes);
        CreateQueueResult result = sqsClient.createQueue(request);
        queueUrl = result.getQueueUrl();
        log.debug("New queue available at: " + queueUrl);
    } else {
        log.debug("Queue already exists: " + queueUrl);
    }

    resolveQueueArn();
}