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

@Test
public void testSend() throws Exception {
    BasicSQSMsgSender msgSender =
        new BasicSQSMsgSender("https://some-fake/url", "us-east-1", "someAwsAccessKey", "someAwsSecretKey");

    Channel mockChannel = mock(Channel.class);
    Event mockEvent = mock(Event.class);
    when(mockEvent.getBody()).thenReturn("This is a test event message".getBytes());
    when(mockChannel.take()).thenReturn(mockEvent);

    AmazonSQS mockSqs = mock(AmazonSQS.class);
    when(mockSqs.sendMessage(any(SendMessageRequest.class))).thenReturn(new SendMessageResult());
    msgSender.setAmazonSQS(mockSqs);

    int eventCount = msgSender.send(mockChannel);
    assertEquals(1, eventCount);
}

@Test
public void testSendEventWithEmptyBody() throws Exception {
    BasicSQSMsgSender msgSender =
        new BasicSQSMsgSender("https://some-fake/url", "us-east-1", "someAwsAccessKey", "someAwsSecretKey");

    Channel mockChannel = mock(Channel.class);
    Event mockEvent = mock(Event.class);
    when(mockEvent.getBody()).thenReturn("".getBytes());
    when(mockChannel.take()).thenReturn(mockEvent);

    AmazonSQS mockSqs = mock(AmazonSQS.class);
    when(mockSqs.sendMessage(any(SendMessageRequest.class))).thenReturn(new SendMessageResult());
    msgSender.setAmazonSQS(mockSqs);

    int eventCount = msgSender.send(mockChannel);
    assertEquals(0, eventCount);
}

@Test
public void testShouldGetFailedResponseAfterSendingTheEvent() {
    final GetQueueUrlResult getQueueUrlResult = new GetQueueUrlResult();
    getQueueUrlResult.setQueueUrl(randomAlphabetic(10));

    final SendMessageResult sendMessageResult = new SendMessageResult();

    final SdkHttpMetadata responseMetadata = mock(SdkHttpMetadata.class);
    when(responseMetadata.getHttpStatusCode()).thenReturn(400);
    sendMessageResult.setSdkHttpMetadata(responseMetadata);

    when(amazonSQS.sendMessage(any(SendMessageRequest.class))).thenThrow(new RuntimeException("expected"));

    assertThatThrownBy(() ->
                                    sqsErrorHandler.onError(randomAlphabetic(10), new RuntimeException(),
                                        EventTypePartition.of(EventType.of(randomAlphabetic(10)),
                                            randomAlphabetic(1)), randomNumeric(10), randomAlphabetic(50)))
        .isInstanceOf(RuntimeException.class).hasMessageContaining("expected");
}

@Test
public void testShouldSendEventToSQS() throws JsonProcessingException {
    final SendMessageResult sendMessageResult = new SendMessageResult();

    final SdkHttpMetadata responseMetadata = mock(SdkHttpMetadata.class);
    when(responseMetadata.getHttpStatusCode()).thenReturn(200);
    sendMessageResult.setSdkHttpMetadata(responseMetadata);

    when(amazonSQS.sendMessage(any(SendMessageRequest.class))).thenReturn(sendMessageResult);

    sqsErrorHandler.onError(randomAlphabetic(10), new RuntimeException(),
        EventTypePartition.of(EventType.of(randomAlphabetic(10)), randomAlphabetic(1)), randomNumeric(10),
        randomAlphabetic(50));

    verify(objectMapper).writeValueAsString(anyString());
    verify(amazonSQS).sendMessage(any(SendMessageRequest.class));
}

public void process(Exchange exchange) throws Exception {
    String body = exchange.getIn().getBody(String.class);
    SendMessageRequest request = new SendMessageRequest(getQueueUrl(), body);
    request.setMessageAttributes(translateAttributes(exchange.getIn().getHeaders(), exchange));
    addDelay(request, exchange);

    LOG.trace("Sending request [{}] from exchange [{}]...", request, exchange);

    SendMessageResult result = getClient().sendMessage(request);

    LOG.trace("Received result [{}]", result);

    Message message = getMessageForResponse(exchange);
    message.setHeader(SqsConstants.MESSAGE_ID, result.getMessageId());
    message.setHeader(SqsConstants.MD5_OF_BODY, result.getMD5OfMessageBody());
}

@Before
public void setup() throws Exception {
    underTest = new SqsProducer(sqsEndpoint);
    sendMessageResult = new SendMessageResult().withMD5OfMessageBody(MESSAGE_MD5).withMessageId(MESSAGE_ID);
    sqsConfiguration = new SqsConfiguration();
    HeaderFilterStrategy headerFilterStrategy = new SqsHeaderFilterStrategy();
    sqsConfiguration.setDelaySeconds(Integer.valueOf(0));
    when(sqsEndpoint.getClient()).thenReturn(amazonSQSClient);
    when(sqsEndpoint.getConfiguration()).thenReturn(sqsConfiguration);
    when(amazonSQSClient.sendMessage(any(SendMessageRequest.class))).thenReturn(sendMessageResult);
    when(exchange.getOut()).thenReturn(outMessage);
    when(exchange.getIn()).thenReturn(inMessage);
    when(exchange.getPattern()).thenReturn(ExchangePattern.InOnly);
    when(inMessage.getBody(String.class)).thenReturn(SAMPLE_MESSAGE_BODY);
    when(sqsEndpoint.getQueueUrl()).thenReturn(QUEUE_URL);
    when(sqsEndpoint.getHeaderFilterStrategy()).thenReturn(headerFilterStrategy);
}

@Override
public SendMessageResult sendMessage(SendMessageRequest sendMessageRequest) throws AmazonServiceException, AmazonClientException {
    Message message = new Message();
    message.setBody(sendMessageRequest.getMessageBody());
    message.setMD5OfBody("6a1559560f67c5e7a7d5d838bf0272ee");
    message.setMessageId("f6fb6f99-5eb2-4be4-9b15-144774141458");
    message.setReceiptHandle("0NNAq8PwvXsyZkR6yu4nQ07FGxNmOBWi5zC9+4QMqJZ0DJ3gVOmjI2Gh/oFnb0IeJqy5Zc8kH4JX7GVpfjcEDjaAPSeOkXQZRcaBqt"
            + "4lOtyfj0kcclVV/zS7aenhfhX5Ixfgz/rHhsJwtCPPvTAdgQFGYrqaHly+etJiawiNPVc=");

    synchronized (messages) {
        messages.add(message);
    }

    SendMessageResult result = new SendMessageResult();
    result.setMessageId("f6fb6f99-5eb2-4be4-9b15-144774141458");
    result.setMD5OfMessageBody("6a1559560f67c5e7a7d5d838bf0272ee");
    return result;
}

@Override
public SendMessageResult sendMessage(AwsParamsDto awsParamsDto, String queueName, String messageText, List<MessageHeader> messageHeaders)
{
    Map<String, MessageAttributeValue> messageAttributes = null;

    if (CollectionUtils.isNotEmpty(messageHeaders))
    {
        messageAttributes = new HashMap<>();

        for (MessageHeader messageHeader : messageHeaders)
        {
            messageAttributes.put(messageHeader.getKey(), new MessageAttributeValue().withDataType("String").withStringValue(messageHeader.getValue()));
        }
    }

    return sqsOperations.sendMessage(queueName, messageText, messageAttributes, awsClientFactory.getAmazonSQSClient(awsParamsDto));
}

@Override
public SendMessageResult sendMessage(String queueName, String messageText, Map<String, MessageAttributeValue> messageAttributes, AmazonSQS amazonSQS)
{
    // Throw a throttling exception for a specific queue name for testing purposes.
    if (queueName.equals(MockAwsOperationsHelper.AMAZON_THROTTLING_EXCEPTION))
    {
        AmazonServiceException throttlingException = new AmazonServiceException("test throttling exception");
        throttlingException.setErrorCode("ThrottlingException");
        throw throttlingException;
    }

    // Throw an illegal state exception for a specific queue name for testing purposes.
    if (queueName.equals(MOCK_SQS_QUEUE_NOT_FOUND_NAME))
    {
        throw new IllegalStateException(String.format("AWS SQS queue with \"%s\" name not found.", queueName));
    }

    // Nothing else to do in the normal case since our unit tests aren't reading messages once they have been published.
    return new SendMessageResult().withMessageId(AbstractDaoTest.MESSAGE_ID);
}

@Test
public void testSendMessage()
{
    // Send an SQS message.
    assertEquals(new SendMessageResult().withMessageId(MESSAGE_ID),
        sqsDao.sendMessage(new AwsParamsDto(), AWS_SQS_QUEUE_NAME, MESSAGE_TEXT, NO_MESSAGE_HEADERS));

    // Send an SQS message using proxy settings.
    assertEquals(new SendMessageResult().withMessageId(MESSAGE_ID), sqsDao
        .sendMessage(new AwsParamsDto(NO_AWS_ACCESS_KEY, NO_AWS_SECRET_KEY, NO_SESSION_TOKEN, HTTP_PROXY_HOST, HTTP_PROXY_PORT), AWS_SQS_QUEUE_NAME,
            MESSAGE_TEXT, NO_MESSAGE_HEADERS));

    // Publish an SQS message with message headers.
    assertEquals(new SendMessageResult().withMessageId(MESSAGE_ID),
        sqsDao.sendMessage(new AwsParamsDto(), AWS_SNS_TOPIC_ARN, MESSAGE_TEXT, Collections.singletonList(new MessageHeader(KEY, VALUE))));
}

@Override
public void write(byte[] bytes) throws IOException {
  if (bytes == null || bytes.length == 0) {
    return;
  }

  final String msg = new String(bytes);

  if (bytes.length > maxMessageSizeInKB * 1024) {
    addWarn(format("Logging event '%s' exceeds the maximum size of %dkB",
        msg, maxMessageSizeInKB));
    return;
  }

  sqs.sendMessageAsync(new SendMessageRequest(queueUrl, msg),
      new AsyncHandler<SendMessageRequest, SendMessageResult>() {
        public void onError(Exception exception) {
          addWarn(format("Appender '%s' failed to send logging event '%s' to '%s'", getName(), msg, queueUrl), exception);
        }
        public void onSuccess(SendMessageRequest request, SendMessageResult result) {
          /** noop **/
        }
      });
}

/**
 * Tests a simple send, receive and delete of a message from the queue
 * resource. Asserts the message contents and its associated attributes.
 */
@Test
@Ignore
public void testSendReceiveDelete() throws InterruptedException {
    SendMessageResult sendMessageResult = queue.sendMessage(TEST_MESSAGE);
    assertNotNull(sendMessageResult);
    assertNotNull(sendMessageResult.getMessageId());

    List<Message> messages = waitForMessagesFromQueue(null);

    assertNotNull(messages);
    assertEquals(1, messages.size());
    Message message = messages.get(0);
    assertMessage(TEST_MESSAGE, sendMessageResult.getMessageId(),
            sendMessageResult.getMD5OfMessageBody(), message);

    queue.deleteMessages(new DeleteMessageBatchRequest()
            .withEntries(new DeleteMessageBatchRequestEntry("msg1", message
                    .getReceiptHandle())));
}

@Test
public void sendMessage_validTextMessage_returnsTrue() throws Exception {
    // Arrange
    AmazonSQSAsync amazonSqs = mock(AmazonSQSAsync.class);
    ArgumentCaptor<SendMessageRequest> sendMessageRequestArgumentCaptor = ArgumentCaptor.forClass(SendMessageRequest.class);
    when(amazonSqs.sendMessage(sendMessageRequestArgumentCaptor.capture())).thenReturn(new SendMessageResult());

    Message<String> stringMessage = MessageBuilder.withPayload("message content").build();
    MessageChannel messageChannel = new QueueMessageChannel(amazonSqs, "http://testQueue");

    // Act
    boolean sent = messageChannel.send(stringMessage);

    // Assert
    verify(amazonSqs, only()).sendMessage(any(SendMessageRequest.class));
    assertEquals("message content", sendMessageRequestArgumentCaptor.getValue().getMessageBody());
    assertTrue(sent);
}

@Test
public void sendMessage_withMimeTypeAsStringHeader_shouldPassItAsMessageAttribute() throws Exception {
    // Arrange
    AmazonSQSAsync amazonSqs = mock(AmazonSQSAsync.class);
    QueueMessageChannel messageChannel = new QueueMessageChannel(amazonSqs, "http://testQueue");
    String mimeTypeAsString = new MimeType("test", "plain", Charset.forName("UTF-8")).toString();
    Message<String> message = MessageBuilder.withPayload("Hello").setHeader(MessageHeaders.CONTENT_TYPE, mimeTypeAsString).build();

    ArgumentCaptor<SendMessageRequest> sendMessageRequestArgumentCaptor = ArgumentCaptor.forClass(SendMessageRequest.class);
    when(amazonSqs.sendMessage(sendMessageRequestArgumentCaptor.capture())).thenReturn(new SendMessageResult());

    // Act
    boolean sent = messageChannel.send(message);

    // Assert
    assertTrue(sent);
    assertEquals(mimeTypeAsString, sendMessageRequestArgumentCaptor.getValue().getMessageAttributes().get(MessageHeaders.CONTENT_TYPE).getStringValue());
}

@Test
public void sendMessage_withMimeTypeHeader_shouldPassItAsMessageAttribute() throws Exception {
    // Arrange
    AmazonSQSAsync amazonSqs = mock(AmazonSQSAsync.class);
    QueueMessageChannel messageChannel = new QueueMessageChannel(amazonSqs, "http://testQueue");
    MimeType mimeType = new MimeType("test", "plain", Charset.forName("UTF-8"));
    Message<String> message = MessageBuilder.withPayload("Hello").setHeader(MessageHeaders.CONTENT_TYPE, mimeType).build();

    ArgumentCaptor<SendMessageRequest> sendMessageRequestArgumentCaptor = ArgumentCaptor.forClass(SendMessageRequest.class);
    when(amazonSqs.sendMessage(sendMessageRequestArgumentCaptor.capture())).thenReturn(new SendMessageResult());

    // Act
    boolean sent = messageChannel.send(message);

    // Assert
    assertTrue(sent);
    assertEquals(mimeType.toString(), sendMessageRequestArgumentCaptor.getValue().getMessageAttributes().get(MessageHeaders.CONTENT_TYPE).getStringValue());
}

@Test
public void sendMessage_withStringMessageHeader_shouldBeSentAsQueueMessageAttribute() throws Exception {
    // Arrange
    AmazonSQSAsync amazonSqs = mock(AmazonSQSAsync.class);
    QueueMessageChannel messageChannel = new QueueMessageChannel(amazonSqs, "http://testQueue");
    String headerValue = "Header value";
    String headerName = "MyHeader";
    Message<String> message = MessageBuilder.withPayload("Hello").setHeader(headerName, headerValue).build();

    ArgumentCaptor<SendMessageRequest> sendMessageRequestArgumentCaptor = ArgumentCaptor.forClass(SendMessageRequest.class);
    when(amazonSqs.sendMessage(sendMessageRequestArgumentCaptor.capture())).thenReturn(new SendMessageResult());

    // Act
    boolean sent = messageChannel.send(message);

    // Assert
    assertTrue(sent);
    assertEquals(headerValue, sendMessageRequestArgumentCaptor.getValue().getMessageAttributes().get(headerName).getStringValue());
    assertEquals(MessageAttributeDataTypes.STRING, sendMessageRequestArgumentCaptor.getValue().getMessageAttributes().get(headerName).getDataType());
}

@Test
public void sendMessage_withBinaryMessageHeader_shouldBeSentAsBinaryMessageAttribute() throws Exception {
    // Arrange
    AmazonSQSAsync amazonSqs = mock(AmazonSQSAsync.class);
    QueueMessageChannel messageChannel = new QueueMessageChannel(amazonSqs, "http://testQueue");
    ByteBuffer headerValue = ByteBuffer.wrap("My binary data!".getBytes());
    String headerName = "MyHeader";
    Message<String> message = MessageBuilder.withPayload("Hello").setHeader(headerName, headerValue).build();

    ArgumentCaptor<SendMessageRequest> sendMessageRequestArgumentCaptor = ArgumentCaptor.forClass(SendMessageRequest.class);
    when(amazonSqs.sendMessage(sendMessageRequestArgumentCaptor.capture())).thenReturn(new SendMessageResult());

    // Act
    boolean sent = messageChannel.send(message);

    // Assert
    assertTrue(sent);
    assertEquals(headerValue, sendMessageRequestArgumentCaptor.getValue().getMessageAttributes().get(headerName).getBinaryValue());
    assertEquals(MessageAttributeDataTypes.BINARY, sendMessageRequestArgumentCaptor.getValue().getMessageAttributes().get(headerName).getDataType());
}

@Test
public void sendMessage_withUuidAsId_shouldConvertUuidToString() throws Exception {
    // Arrange
    AmazonSQSAsync amazonSqs = mock(AmazonSQSAsync.class);
    QueueMessageChannel messageChannel = new QueueMessageChannel(amazonSqs, "http://testQueue");
    Message<String> message = MessageBuilder.withPayload("Hello").build();
    UUID uuid = (UUID) message.getHeaders().get(MessageHeaders.ID);

    ArgumentCaptor<SendMessageRequest> sendMessageRequestArgumentCaptor = ArgumentCaptor.forClass(SendMessageRequest.class);
    when(amazonSqs.sendMessage(sendMessageRequestArgumentCaptor.capture())).thenReturn(new SendMessageResult());

    // Act
    boolean sent = messageChannel.send(message);

    // Assert
    assertTrue(sent);
    assertEquals(uuid.toString(), sendMessageRequestArgumentCaptor.getValue().getMessageAttributes().get(MessageHeaders.ID).getStringValue());
}

@Test
@SuppressWarnings("unchecked")
public void sendMessage_withTimeout_sendsMessageAsyncAndReturnsTrueOnceFutureCompleted() throws Exception {
    // Arrange
    Future<SendMessageResult> future = mock(Future.class);
    when(future.get(1000, TimeUnit.MILLISECONDS)).thenReturn(new SendMessageResult());
    AmazonSQSAsync amazonSqs = mock(AmazonSQSAsync.class);
    when(amazonSqs.sendMessageAsync(any(SendMessageRequest.class))).thenReturn(future);
    QueueMessageChannel queueMessageChannel = new QueueMessageChannel(amazonSqs, "http://testQueue");

    // Act
    boolean result = queueMessageChannel.send(MessageBuilder.withPayload("Hello").build(), 1000);

    // Assert
    assertTrue(result);
    verify(amazonSqs, only()).sendMessageAsync(any(SendMessageRequest.class));
}

@Test
@SuppressWarnings("unchecked")
public void sendMessage_withSendMessageAsyncTakingMoreTimeThanSpecifiedTimeout_returnsFalse() throws Exception {
    // Arrange
    Future<SendMessageResult> future = mock(Future.class);
    when(future.get(1000, TimeUnit.MILLISECONDS)).thenThrow(new TimeoutException());
    AmazonSQSAsync amazonSqs = mock(AmazonSQSAsync.class);
    when(amazonSqs.sendMessageAsync(any(SendMessageRequest.class))).thenReturn(future);
    QueueMessageChannel queueMessageChannel = new QueueMessageChannel(amazonSqs, "http://testQueue");

    // Act
    boolean result = queueMessageChannel.send(MessageBuilder.withPayload("Hello").build(), 1000);

    // Assert
    assertFalse(result);
}

@Test
@SuppressWarnings("unchecked")
public void sendMessage_withExecutionExceptionWhileSendingAsyncMessage_throwMessageDeliveryException() throws Exception {
    // Arrange
    Future<SendMessageResult> future = mock(Future.class);
    when(future.get(1000, TimeUnit.MILLISECONDS)).thenThrow(new ExecutionException(new Exception()));
    AmazonSQSAsync amazonSqs = mock(AmazonSQSAsync.class);
    when(amazonSqs.sendMessageAsync(any(SendMessageRequest.class))).thenReturn(future);
    QueueMessageChannel queueMessageChannel = new QueueMessageChannel(amazonSqs, "http://testQueue");

    // Assert
    this.expectedException.expect(MessageDeliveryException.class);

    // Act
    queueMessageChannel.send(MessageBuilder.withPayload("Hello").build(), 1000);

}

@Test
public void sendMessage_withDelayHeader_shouldSetDelayOnSendMessageRequestAndNotSetItAsHeaderAsMessageAttribute() throws Exception {
    // Arrange
    AmazonSQSAsync amazonSqs = mock(AmazonSQSAsync.class);

    ArgumentCaptor<SendMessageRequest> sendMessageRequestArgumentCaptor = ArgumentCaptor.forClass(SendMessageRequest.class);
    when(amazonSqs.sendMessage(sendMessageRequestArgumentCaptor.capture())).thenReturn(new SendMessageResult());

    QueueMessageChannel queueMessageChannel = new QueueMessageChannel(amazonSqs, "http://testQueue");
    Message<String> message = MessageBuilder.withPayload("Hello").setHeader(SqsMessageHeaders.SQS_DELAY_HEADER, 15).build();

    // Act
    queueMessageChannel.send(message);

    // Assert
    SendMessageRequest sendMessageRequest = sendMessageRequestArgumentCaptor.getValue();
    assertEquals(new Integer(15), sendMessageRequest.getDelaySeconds());
    assertFalse(sendMessageRequest.getMessageAttributes().containsKey(SqsMessageHeaders.SQS_DELAY_HEADER));
}

@Test
public void sendMessage_withoutDelayHeader_shouldNotSetDelayOnSendMessageRequestAndNotSetHeaderAsMessageAttribute() throws Exception {
    // Arrange
    AmazonSQSAsync amazonSqs = mock(AmazonSQSAsync.class);

    ArgumentCaptor<SendMessageRequest> sendMessageRequestArgumentCaptor = ArgumentCaptor.forClass(SendMessageRequest.class);
    when(amazonSqs.sendMessage(sendMessageRequestArgumentCaptor.capture())).thenReturn(new SendMessageResult());

    QueueMessageChannel queueMessageChannel = new QueueMessageChannel(amazonSqs, "http://testQueue");
    Message<String> message = MessageBuilder.withPayload("Hello").build();

    // Act
    queueMessageChannel.send(message);

    // Assert
    SendMessageRequest sendMessageRequest = sendMessageRequestArgumentCaptor.getValue();
    assertNull(sendMessageRequest.getDelaySeconds());
    assertFalse(sendMessageRequest.getMessageAttributes().containsKey(SqsMessageHeaders.SQS_DELAY_HEADER));
}

@Test
public void sendMessage_withGroupIdHeader_shouldSetGroupIdOnSendMessageRequestAndNotSetItAsHeaderAsMessageAttribute() throws Exception {
    // Arrange
    AmazonSQSAsync amazonSqs = mock(AmazonSQSAsync.class);

    ArgumentCaptor<SendMessageRequest> sendMessageRequestArgumentCaptor = ArgumentCaptor.forClass(SendMessageRequest.class);
    when(amazonSqs.sendMessage(sendMessageRequestArgumentCaptor.capture())).thenReturn(new SendMessageResult());

    QueueMessageChannel queueMessageChannel = new QueueMessageChannel(amazonSqs, "http://testQueue");
    Message<String> message = MessageBuilder.withPayload("Hello").setHeader(SqsMessageHeaders.SQS_GROUP_ID_HEADER, "id-5").build();

    // Act
    queueMessageChannel.send(message);

    // Assert
    SendMessageRequest sendMessageRequest = sendMessageRequestArgumentCaptor.getValue();
    assertEquals("id-5", sendMessageRequest.getMessageGroupId());
    assertFalse(sendMessageRequest.getMessageAttributes().containsKey(SqsMessageHeaders.SQS_GROUP_ID_HEADER));
}

@Test
public void sendMessage_withDeduplicationIdHeader_shouldSetDeduplicationIdOnSendMessageRequestAndNotSetItAsHeaderAsMessageAttribute() throws Exception {
    // Arrange
    AmazonSQSAsync amazonSqs = mock(AmazonSQSAsync.class);

    ArgumentCaptor<SendMessageRequest> sendMessageRequestArgumentCaptor = ArgumentCaptor.forClass(SendMessageRequest.class);
    when(amazonSqs.sendMessage(sendMessageRequestArgumentCaptor.capture())).thenReturn(new SendMessageResult());

    QueueMessageChannel queueMessageChannel = new QueueMessageChannel(amazonSqs, "http://testQueue");
    Message<String> message = MessageBuilder.withPayload("Hello").setHeader(SqsMessageHeaders.SQS_DEDUPLICATION_ID_HEADER, "id-5").build();

    // Act
    queueMessageChannel.send(message);

    // Assert
    SendMessageRequest sendMessageRequest = sendMessageRequestArgumentCaptor.getValue();
    assertEquals("id-5", sendMessageRequest.getMessageDeduplicationId());
    assertFalse(sendMessageRequest.getMessageAttributes().containsKey(SqsMessageHeaders.SQS_DEDUPLICATION_ID_HEADER));
}

/**
 * Test sendInternal input with SQSTextMessage
 */
@Test
public void testSendInternalSQSTextMessage() throws JMSException {

    String messageBody = "MyText1";
    SQSTextMessage msg = spy(new SQSTextMessage(messageBody));
    msg.setStringProperty(SQSMessagingClientConstants.JMSX_GROUP_ID, GROUP_ID);
    msg.setStringProperty(SQSMessagingClientConstants.JMS_SQS_DEDUPLICATION_ID, DEDUP_ID);

    when(amazonSQSClient.sendMessage(any(SendMessageRequest.class)))
            .thenReturn(new SendMessageResult().withMessageId(MESSAGE_ID).withSequenceNumber(SEQ_NUMBER));

    producer.sendInternal(destination, msg);

    verify(amazonSQSClient).sendMessage(argThat(new sendMessageRequestMatcher(QUEUE_URL, messageBody, SQSMessage.TEXT_MESSAGE_TYPE, GROUP_ID, DEDUP_ID)));
    verify(msg).setJMSDestination(destination);
    verify(msg).setJMSMessageID("ID:" + MESSAGE_ID);
    verify(msg).setSQSMessageId(MESSAGE_ID);
    verify(msg).setSequenceNumber(SEQ_NUMBER);
}

/**
 * Test sendInternal input with SQSObjectMessage
 */
@Test
public void testSendInternalSQSObjectMessage() throws JMSException {

    HashSet<String> set = new HashSet<String>();
    set.add("data1");

    SQSObjectMessage msg = spy(new SQSObjectMessage(set));
    msg.setStringProperty(SQSMessagingClientConstants.JMSX_GROUP_ID, GROUP_ID);
    msg.setStringProperty(SQSMessagingClientConstants.JMS_SQS_DEDUPLICATION_ID, DEDUP_ID);
    String msgBody = msg.getMessageBody();

    when(amazonSQSClient.sendMessage(any(SendMessageRequest.class)))
            .thenReturn(new SendMessageResult().withMessageId(MESSAGE_ID).withSequenceNumber(SEQ_NUMBER));

    producer.sendInternal(destination, msg);

    verify(amazonSQSClient).sendMessage(argThat(new sendMessageRequestMatcher(QUEUE_URL, msgBody, SQSMessage.OBJECT_MESSAGE_TYPE, GROUP_ID, DEDUP_ID)));
    verify(msg).setJMSDestination(destination);
    verify(msg).setJMSMessageID("ID:" + MESSAGE_ID);
    verify(msg).setSQSMessageId(MESSAGE_ID);
    verify(msg).setSequenceNumber(SEQ_NUMBER);
}

/**
 * Test sendInternal input with SQSByteMessage
 */
@Test
public void testSendInternalSQSByteMessage() throws JMSException {

    SQSBytesMessage msg = spy(new SQSBytesMessage());
    msg.setStringProperty(SQSMessagingClientConstants.JMSX_GROUP_ID, GROUP_ID);
    msg.setStringProperty(SQSMessagingClientConstants.JMS_SQS_DEDUPLICATION_ID, DEDUP_ID);
    msg.writeByte((byte)0);
    msg.reset();

    when(amazonSQSClient.sendMessage(any(SendMessageRequest.class)))
            .thenReturn(new SendMessageResult().withMessageId(MESSAGE_ID).withSequenceNumber(SEQ_NUMBER));

    producer.sendInternal(destination, msg);

    String messageBody = "AA==";
    verify(amazonSQSClient).sendMessage(argThat(new sendMessageRequestMatcher(QUEUE_URL, messageBody, SQSMessage.BYTE_MESSAGE_TYPE, GROUP_ID, DEDUP_ID)));

    verify(msg).setJMSDestination(destination);
    verify(msg).setJMSMessageID("ID:" + MESSAGE_ID);
    verify(msg).setSQSMessageId(MESSAGE_ID);
    verify(msg).setSequenceNumber(SEQ_NUMBER);
}

public void client2CanReceiveTwiceAfterInitialEmpty() {
    final String queueUrl = someNewQueue();
    final ReceiveMessageResult result1 = _sqs2.receiveMessage(new ReceiveMessageRequest(queueUrl).withWaitTimeSeconds(1).withMaxNumberOfMessages(1));
    Assert.assertEquals(result1.getMessages().size(), 0);

    final SendMessageResult sendResult1 = _sqs1.sendMessage(new SendMessageRequest(queueUrl, someMessageBody()));
    final ReceiveMessageResult result2 = _sqs2.receiveMessage(new ReceiveMessageRequest(queueUrl).
            withWaitTimeSeconds(10).
            withMaxNumberOfMessages(1).
            withVisibilityTimeout(60));
    Assert.assertEquals(result2.getMessages().size(), 1, "first receive failed");

    final SendMessageResult sendResult2 = _sqs1.sendMessage(new SendMessageRequest(queueUrl, someMessageBody()));
    final ReceiveMessageResult result3 = _sqs2.receiveMessage(new ReceiveMessageRequest(queueUrl).
            withWaitTimeSeconds(20).
            withMaxNumberOfMessages(1));
    Assert.assertEquals(result3.getMessages().size(), 1, "second receive failed");
}

public void client1GetsFromBoth() {
    final String queueUrl = someNewQueue();
    final SendMessageResult sendResult1 = _sqs1.sendMessage(new SendMessageRequest(queueUrl, someMessageBody()));
    final SendMessageResult sendResult2 = _sqs2.sendMessage(new SendMessageRequest(queueUrl, someMessageBody()));

    final ReceiveMessageResult receiveMessageResult1 = _sqs1.receiveMessage(new ReceiveMessageRequest(queueUrl).
            withMaxNumberOfMessages(1).
            withWaitTimeSeconds(20));

    Assert.assertEquals(receiveMessageResult1.getMessages().size(), 1);

    final ReceiveMessageResult receiveMessageResult2 = _sqs1.receiveMessage(new ReceiveMessageRequest(queueUrl).
            withWaitTimeSeconds(20));

    Assert.assertEquals(receiveMessageResult2.getMessages().size(), 1);
}

/**
 * Executes the outbound Sqs Operation.
 * 
 */
public Object executeOutboundOperation(final Message<?> message) {

    try {
        String serializedMessage = messageMarshaller.serialize(message);
        if (queue == null) {
            SendMessageRequest request = new SendMessageRequest(queueUrl,
                    serializedMessage);
            SendMessageResult result = sqsClient.sendMessage(request);
            log.debug("Message sent, Id:" + result.getMessageId());
        } else {
            queue.add(serializedMessage);
        }
    } catch (MessageMarshallerException e) {
        log.error(e.getMessage(), e);
        throw new MessagingException(e.getMessage(), e.getCause());
    }

    return message.getPayload();
}

@Override
public Either<HttpFailure,SendMessageResult> sendMessage(final URI queueURI,
                                                            final String message) {
    return new AwsSQSHttpClosure<SendMessageResult>(client_, SC_OK,
        new SendMessageResultStaxUnmarshaller()) {
        @Override
        public void validate() throws Exception {
            checkNotNull(queueURI, "Queue URI cannot be null.");
            checkNotNull(message, "Message to send cannot be null.");
        }
        @Override
        public void prepare(final AwsHttpRequest request) throws Exception {
            request.addParameter(SQS_ACTION_PARAM, SQS_ACTION_SEND_MESSAGE);
            request.addParameter(SQS_MESSAGE_BODY_PARAM, message);
        }
    }.post(queueURI);
}

private Mono<SendMessageResult> sendMessage(String body) {

        SendMessageRequest sendMessageRequest = new SendMessageRequest(
                SQS_ENDPOINT,
                body
        );

        return Mono.fromFuture(
                Utils.makeCompletableFuture(
                        sqs.sendMessageAsync(sendMessageRequest)))
                .doOnError((throwable -> LOG.error(Utils.error.failed_sqs, body)));
    }

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

@Override protected void doEnqueue(Callback<Void> callback) {
  future = get().sendMessageAsync(message,
      new AsyncHandler<SendMessageRequest, SendMessageResult>() {
        @Override public void onError(Exception e) {
          callback.onError(e);
        }

        @Override
        public void onSuccess(SendMessageRequest request, SendMessageResult sendMessageResult) {
          callback.onSuccess(null);
        }
      });
  if (future.isCancelled()) throw new IllegalStateException("cancelled sending spans");
}

@Test
public void testSendEmptyChannel() throws Exception {
    BasicSQSMsgSender msgSender =
        new BasicSQSMsgSender("https://some-fake/url", "us-east-1", "someAwsAccessKey", "someAwsSecretKey");

    Channel mockChannel = mock(Channel.class);
    when(mockChannel.take()).thenReturn(null);

    AmazonSQS mockSqs = mock(AmazonSQS.class);
    when(mockSqs.sendMessage(any(SendMessageRequest.class))).thenReturn(new SendMessageResult());
    msgSender.setAmazonSQS(mockSqs);

    int eventCount = msgSender.send(mockChannel);
    assertEquals(0, eventCount);
}

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

@Override
public void add(JSONObject json) throws Exception {
    try {
        SendMessageRequest sendMessageRequest = new SendMessageRequest()
        .withQueueUrl(this.queueUrl)
        .withMessageBody(json.toString());
        SendMessageResult sendMessageResult = this.client.sendMessage(sendMessageRequest);
        lastMessageId = sendMessageResult.getMessageId();
    } catch (Throwable e){
        throw new Exception(e);
    }
}

public SendMessageRequestHolder(
        SendMessageRequest sendMessageRequest,
        Future<SendMessageResult> sendMessageResultFuture,
        int tries) {
    this.sendMessageRequest = sendMessageRequest;
    this.sendMessageResultFuture = sendMessageResultFuture;
    this.tries = tries;
}