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

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

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

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

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

@Test
public void testAssertQueueAlreadyExistsAndMatches() {
    when(requestSenderMock.sendRequest(any(CreateQueueAction.class))).thenReturn(Single.just(
            new CreateQueueResult().withQueueUrl(QUEUE_URL)
    ));
    SqsQueue<String> queue = client.upsertQueue(QUEUE_CONFIG).blockingGet();
    assertThat(queue.getQueueUrl()).isEqualTo(QUEUE_URL);
    verify(requestSenderMock).sendRequest(any(CreateQueueAction.class));
}

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

public Scanner(ScanConfig config) {
    opt = config;
    ProfileCredentialsProvider creds = new ProfileCredentialsProvider(opt.profile());
    creds.getCredentials(); // credible credential criteria
    s3 = new AmazonS3Client(creds);
    sqs = new AmazonSQSClient(creds);

    CreateQueueResult queueResult = sqs.createQueue(opt.queue());
    queueUrl = queueResult.getQueueUrl();

    filter = Pattern.compile(opt.filter(), Pattern.CASE_INSENSITIVE);
    max = Long.parseLong(opt.max());
}

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

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

public void produceMsg(String[] msgs, boolean purgeFirst) throws Exception
{
  CreateQueueResult res = sqs.createQueue(getCurrentQueueName());
  if (purgeFirst) {
    PurgeQueueRequest purgeReq = new PurgeQueueRequest(res.getQueueUrl());
    sqs.purgeQueue(purgeReq);
  }
  for (String text : msgs) {
    sqs.sendMessage(res.getQueueUrl(), text);
  }
}

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

/**
 * Note here we attempt to the TOCQueue which may take some time to be shown as available
 * @param isConsumer
 * @param maxAttempts
 * @throws Exception
 */
public void connectToQueue(boolean isConsumer, int maxAttempts) throws Exception{

    for (int i=0; i<maxAttempts; i++) {

        logger.debug("connectToQueue() attempt: " + (i+1));

        ListQueuesResult queuesResult = sqsClient.listQueues();
        if (queuesResult != null) {
            for (String queueUrl : queuesResult.getQueueUrls()) {
                if (queueUrl.indexOf(sqsQueueName) != -1) {
                    tocQueueUrl = queueUrl;
                    break;
                }
            }
        }

        // if consumer, retry, otherwise is master, so just exit quick to create...
        if (tocQueueUrl == null && isConsumer) {
            Thread.currentThread().sleep(1000);
            continue;
        } else {
            break; // exit;
        }
    }

    if (tocQueueUrl == null && !isConsumer) {
        CreateQueueResult createQueueResult = sqsClient.createQueue(sqsQueueName);
        this.tocQueueUrl = createQueueResult.getQueueUrl();

    } else if (tocQueueUrl == null) {
        throw new Exception("TOCQueue() isConsumer:"+ isConsumer+ " cannot start, sqsQueueName has yet to be created by master?: " + sqsQueueName);
    }
}

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

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

@Override
public Either<HttpFailure,CreateQueueResult> createQueue(final String queueName,
                                                            final Integer defaultVisibilityTimeout) {
    return new AwsSQSHttpClosure<CreateQueueResult>(client_, SC_OK,
        new CreateQueueResultStaxUnmarshaller()) {
        @Override
        public void validate() throws Exception {
            checkNotNull(queueName, "Queue name cannot be null.");
            checkState(isValidQueueName(queueName), "Invalid queue name, " +
                "did not match expected queue name pattern.");
            if(defaultVisibilityTimeout != null) {
                checkState(defaultVisibilityTimeout <= SQS_MAX_VISIBILITY_TIMEOUT,
                    "Default visibility timeout cannot be greater than: " +
                    SQS_MAX_VISIBILITY_TIMEOUT);
            }
        }
        @Override
        public void prepare(final AwsHttpRequest request) throws Exception {
            request.addParameter(SQS_ACTION_PARAM, SQS_ACTION_CREATE_QUEUE);
            request.addParameter(SQS_QUEUE_NAME_PARAM, queueName);
            if(defaultVisibilityTimeout != null) {
                request.addParameter(SQS_DEFAULT_VISIBILITY_TIMEOUT_PARAM,
                    Integer.toString(defaultVisibilityTimeout));
            }
        }
    }.post();
}

@Test
public void testSendChangeVisibilityReceiveDeleteMessage_shouldSendChangeVisibilityReceiveAndDeleteMessage() {
  // create queue
  CreateQueueResult createdQueue = sqs.createQueue(new CreateQueueRequest().withQueueName("tea-earl-grey-queue"));
  // send message
  String messageBody = "{\"life-universe-everything\":42}";
  SendMessageResult sendResult = sqs.sendMessage(new SendMessageRequest().withDelaySeconds(0).withMessageBody(messageBody)
      .withMessageGroupId("some-group-id-123").withQueueUrl(createdQueue.getQueueUrl()));
  assertNotNull("message sending returned ok", sendResult);
  assertNotNull("verify body MD5 exists",sendResult.getMD5OfMessageBody());
  assertNotNull("verify message id exists",sendResult.getMessageId());
  // receive message
  ReceiveMessageResult messageResult = sqs.receiveMessage(new ReceiveMessageRequest()
      .withMaxNumberOfMessages(3).withQueueUrl(createdQueue.getQueueUrl()).withVisibilityTimeout(10)
      .withWaitTimeSeconds(0));
  assertNotNull("verify received message returned ok",messageResult);
  assertEquals("verify correct receive count", 1, messageResult.getMessages().size());
  Message firstMessage = messageResult.getMessages().get(0);
  assertEquals("verify correct body returned",messageBody,firstMessage.getBody());
  assertEquals("verify correct message MD5",getAwsMessageMD5(messageBody),firstMessage.getMD5OfBody());
  assertNotNull("verify message id exists",firstMessage.getMessageId());
  assertNotNull("verify receipt handle exists",firstMessage.getReceiptHandle());

  // extend visibility timeout
  ChangeMessageVisibilityResult visibilityResult = sqs.changeMessageVisibility(new ChangeMessageVisibilityRequest()
      .withQueueUrl(createdQueue.getQueueUrl()).withReceiptHandle(firstMessage.getReceiptHandle()).withVisibilityTimeout(40));
  assertNotNull("changing visibility returned ok", visibilityResult);

  // verify if message is invisible
  ReceiveMessageResult emptyResult = sqs.receiveMessage(new ReceiveMessageRequest()
      .withMaxNumberOfMessages(1).withQueueUrl(createdQueue.getQueueUrl()).withVisibilityTimeout(20)
      .withWaitTimeSeconds(0));
  assertTrue("at visibility timeout the message should not be available.", emptyResult.getMessages().isEmpty());

  // delete message from queue
  DeleteMessageResult deleteResult = sqs.deleteMessage(new DeleteMessageRequest()
      .withQueueUrl(createdQueue.getQueueUrl()).withReceiptHandle(firstMessage.getReceiptHandle()));
  assertNotNull("verify deletion returned ok",deleteResult);

  assertTrue("queue must be empty after removal",getQueues().get("tea-earl-grey-queue").getMessageQueue().isEmpty());
  assertTrue("invisibility-queue must be empty after removal",getQueues().get("tea-earl-grey-queue").getInvisibilityQueueFor(firstMessage.getReceiptHandle()).isEmpty());

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

@Test
public void testBulkSendDelete_shouldWork() {
  // create queue
  CreateQueueResult createdQueue = sqs.createQueue(new CreateQueueRequest().withQueueName("tea-earl-grey-queue"));
  // send batch
  SendMessageBatchRequestEntry firstRequest = new SendMessageBatchRequestEntry().withDelaySeconds(0).withId("one")
      .withMessageGroupId("groupee").withMessageBody("{\"XOXO\":234}");
  SendMessageBatchRequestEntry secondRequest = new SendMessageBatchRequestEntry().withDelaySeconds(0).withId("two")
      .withMessageGroupId("groupee").withMessageBody("{\"Quinoa\":\"Readymade\",\"vegan\":true}");
  SendMessageBatchRequestEntry thirdRequest = new SendMessageBatchRequestEntry().withDelaySeconds(0).withId("three")
      .withMessageGroupId("groupee").withMessageBody("{\"VHS\":\"street art slow-carb\"}");
  // verify send batch result
  SendMessageBatchResult sendResult = sqs.sendMessageBatch(new SendMessageBatchRequest().withQueueUrl(createdQueue.getQueueUrl())
      .withEntries(ImmutableList.of(firstRequest,secondRequest, thirdRequest)));
  assertNotNull("verify that batch send returned ok", sendResult);
  assertTrue("no request failed",sendResult.getFailed().isEmpty());
  assertEquals("verify successfull message count", 3, sendResult.getSuccessful().size());
  SendMessageBatchResultEntry firstResultEntry = sendResult.getSuccessful().stream().filter(msg -> msg.getId().equals("one")).findAny().get();
  assertEquals("verify correct message MD5",getAwsMessageMD5("{\"XOXO\":234}"),firstResultEntry.getMD5OfMessageBody());
  assertNotNull("verify message id exists",firstResultEntry.getMessageId());

  ReceiveMessageResult receivedMessagesResult = sqs.receiveMessage(new ReceiveMessageRequest().withQueueUrl(createdQueue.getQueueUrl()).withMaxNumberOfMessages(4));
  // delete batch
  List<DeleteMessageBatchRequestEntry> deleteRequests = new ArrayList<>();
  deleteRequests.add(new DeleteMessageBatchRequestEntry().withId("one").withReceiptHandle(receivedMessagesResult.getMessages().get(0).getReceiptHandle()));
  deleteRequests.add(new DeleteMessageBatchRequestEntry().withId("two").withReceiptHandle(receivedMessagesResult.getMessages().get(0).getReceiptHandle()));
  deleteRequests.add(new DeleteMessageBatchRequestEntry().withId("three").withReceiptHandle(receivedMessagesResult.getMessages().get(0).getReceiptHandle())); 

  DeleteMessageBatchResult deleteBatchResult = sqs.deleteMessageBatch(new DeleteMessageBatchRequest().withQueueUrl(createdQueue.getQueueUrl()).withEntries(deleteRequests));

  // verify delete batch result
  assertNotNull("verify that batch delete returned ok", deleteBatchResult);
  assertTrue("no request failed",deleteBatchResult.getFailed().isEmpty());
  assertEquals("verify successfull message count", 3, deleteBatchResult.getSuccessful().size());
  assertTrue("queue must be empty after removal",getQueues().get("tea-earl-grey-queue").getMessageQueue().isEmpty());
  for(Message message : receivedMessagesResult.getMessages()) {
    assertTrue("invisibility-queue must be empty after removal",getQueues().get("tea-earl-grey-queue").getInvisibilityQueueFor(message.getReceiptHandle()).isEmpty());
  }

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

public Observable<CreateQueueResult> createQueueAsync(CreateQueueRequest request) {
    return Observable.from(sqsClient.createQueueAsync(request));
}

public Observable<CreateQueueResult> createQueueAsync(String queueName) {
    return Observable.from(sqsClient.createQueueAsync(queueName));
}

@Override
public CreateQueueResult createQueue(CreateQueueRequest createQueueRequest) throws AmazonServiceException, AmazonClientException {
    throw new AmazonServiceException("forced exception for test if this method is called");
}

public String createMessageQueue(String queueName) {
    CreateQueueResult createQueueResult = sqsClient.createQueue(queueName);
    return createQueueResult.getQueueUrl();
}

@Override
public Queue createQueue(String queueName) {
    return createQueue(queueName, (ResultCapture<CreateQueueResult>)null);
}

private String someNewQueue() {
    final String queueName = someQueueName();
    final CreateQueueResult queue = _amazonSQS.createQueue(new CreateQueueRequest(queueName));
    return queue.getQueueUrl();
}

private String someNewQueue() {
    final String queueName = someQueueName();
    final CreateQueueResult queue = _sqs1.createQueue(new CreateQueueRequest(queueName));
    return queue.getQueueUrl();
}

private String someNewQueue(String queueName) {
    final CreateQueueResult queue = _amazonSQS1.createQueue(new CreateQueueRequest(queueName));
    return queue.getQueueUrl();
}

public String createQueue(String name){
    CreateQueueRequest request=new CreateQueueRequest();
    request.setQueueName(name);
    CreateQueueResult res= sqsClient.createQueue(request);
    return res.getQueueUrl();
}

@Override
public Either<HttpFailure,CreateQueueResult> createQueue(final String queueName) {
    return createQueue(queueName, null);
}

/**
 * Calls <code>createQueue</code> to create the queue with the provided queue attributes
 * if any, and wraps <code>AmazonClientException</code>
 * 
 * @param createQueueRequest
 *            Container for the necessary parameters to execute the
 *            createQueue service method on AmazonSQS.
 * @return The response from the createQueue service method, as returned by
 *         AmazonSQS. This call creates a new queue, or returns the URL of
 *         an existing one.
 * @throws JMSException
 */
public CreateQueueResult createQueue(CreateQueueRequest createQueueRequest) throws JMSException {
    try {
        prepareRequest(createQueueRequest);
        return amazonSQSClient.createQueue(createQueueRequest);
    } catch (AmazonClientException e) {
        throw handleException(e, "createQueue");   
    }    
}

/**
 * <p>
 * Creates a new queue, or returns the URL of an existing one. When you
 * request <code>CreateQueue</code> , you provide a name for the queue. To
 * successfully create a new queue, you must provide a name that is unique
 * within the scope of your own queues.
 * </p>
 * <p>
 * <b>NOTE:</b> If you delete a queue, you must wait at least 60 seconds
 * before creating a queue with the same name.
 * </p>
 * <p>
 * You may pass one or more attributes in the request. If you do not provide
 * a value for any attribute, the queue will have the default value for that
 * attribute. Permitted attributes are the same that can be set using
 * SetQueueAttributes.
 * </p>
 * <p>
 * <b>NOTE:</b> Use GetQueueUrl to get a queue's URL. GetQueueUrl requires
 * only the QueueName parameter.
 * </p>
 * <p>
 * If you provide the name of an existing queue, along with the exact names
 * and values of all the queue's attributes, <code>CreateQueue</code>
 * returns the queue URL for the existing queue. If the queue name,
 * attribute names, or attribute values do not match an existing queue,
 * <code>CreateQueue</code> returns an error.
 * </p>
 * <p>
 * <b>NOTE:</b>Some API actions take lists of parameters. These lists are
 * specified using the param.n notation. Values of n are integers starting
 * from 1. For example, a parameter list with two elements looks like this:
 * </p>
 * <p>
 * <code>&Attribute.1=this</code>
 * </p>
 * <p>
 * <code>&Attribute.2=that</code>
 * </p>
 *
 * @param createQueueRequest
 *            Container for the necessary parameters to execute the
 *            CreateQueue service method on AmazonSQS.
 * 
 * @return The response from the CreateQueue service method, as returned by
 *         AmazonSQS.
 * 
 * @throws QueueNameExistsException
 * @throws QueueDeletedRecentlyException
 *
 * @throws AmazonClientException
 *             If any internal errors are encountered inside the client
 *             while attempting to make the request or handle the response.
 *             For example if a network connection is not available.
 * @throws AmazonServiceException
 *             If an error response is returned by AmazonSQS indicating
 *             either a problem with the data in the request, or a server
 *             side issue.
 */
public CreateQueueResult createQueue(CreateQueueRequest createQueueRequest) throws AmazonServiceException,
        AmazonClientException {

    return amazonSqsToBeExtended.createQueue(createQueueRequest);
}

/**
 * <p>
 * Creates a new queue, or returns the URL of an existing one. When you
 * request <code>CreateQueue</code> , you provide a name for the queue. To
 * successfully create a new queue, you must provide a name that is unique
 * within the scope of your own queues.
 * </p>
 * <p>
 * <b>NOTE:</b> If you delete a queue, you must wait at least 60 seconds
 * before creating a queue with the same name.
 * </p>
 * <p>
 * You may pass one or more attributes in the request. If you do not provide
 * a value for any attribute, the queue will have the default value for that
 * attribute. Permitted attributes are the same that can be set using
 * SetQueueAttributes.
 * </p>
 * <p>
 * <b>NOTE:</b> Use GetQueueUrl to get a queue's URL. GetQueueUrl requires
 * only the QueueName parameter.
 * </p>
 * <p>
 * If you provide the name of an existing queue, along with the exact names
 * and values of all the queue's attributes, <code>CreateQueue</code>
 * returns the queue URL for the existing queue. If the queue name,
 * attribute names, or attribute values do not match an existing queue,
 * <code>CreateQueue</code> returns an error.
 * </p>
 * <p>
 * <b>NOTE:</b>Some API actions take lists of parameters. These lists are
 * specified using the param.n notation. Values of n are integers starting
 * from 1. For example, a parameter list with two elements looks like this:
 * </p>
 * <p>
 * <code>&Attribute.1=this</code>
 * </p>
 * <p>
 * <code>&Attribute.2=that</code>
 * </p>
 * 
 * @param queueName
 *            The name for the queue to be created.
 * 
 * @return The response from the CreateQueue service method, as returned by
 *         AmazonSQS.
 * 
 * @throws QueueNameExistsException
 * @throws QueueDeletedRecentlyException
 *
 * @throws AmazonClientException
 *             If any internal errors are encountered inside the client
 *             while attempting to make the request or handle the response.
 *             For example if a network connection is not available.
 * @throws AmazonServiceException
 *             If an error response is returned by AmazonSQS indicating
 *             either a problem with the data in the request, or a server
 *             side issue.
 */
public CreateQueueResult createQueue(String queueName) throws AmazonServiceException, AmazonClientException {

    return amazonSqsToBeExtended.createQueue(queueName);
}

/**
 * Performs the <code>CreateQueue</code> action and use a ResultCapture to
 * retrieve the low-level client response.
 *
 * <p>
 *
 * @return The <code>Queue</code> resource object associated with the result
 *         of this action.
 * @see CreateQueueRequest
 */
com.amazonaws.resources.sqs.Queue createQueue(CreateQueueRequest request,
        ResultCapture<CreateQueueResult> extractor);

/**
 * The convenient method form for the <code>CreateQueue</code> action.
 *
 * @see #createQueue(CreateQueueRequest, ResultCapture)
 */
com.amazonaws.resources.sqs.Queue createQueue(String queueName,
        ResultCapture<CreateQueueResult> extractor);