import com.amazonaws.services.sqs.model.MessageAttributeValue; //导入依赖的package包/类
private Exchange createExchange(ExchangePattern pattern, com.amazonaws.services.sqs.model.Message msg) {
    Exchange exchange = super.createExchange(pattern);
    Message message = exchange.getIn();
    message.setBody(msg.getBody());
    message.setHeaders(new HashMap<String, Object>(msg.getAttributes()));
    message.setHeader(SqsConstants.MESSAGE_ID, msg.getMessageId());
    message.setHeader(SqsConstants.MD5_OF_BODY, msg.getMD5OfBody());
    message.setHeader(SqsConstants.RECEIPT_HANDLE, msg.getReceiptHandle());
    message.setHeader(SqsConstants.ATTRIBUTES, msg.getAttributes());
    message.setHeader(SqsConstants.MESSAGE_ATTRIBUTES, msg.getMessageAttributes());
    
    //Need to apply the SqsHeaderFilterStrategy this time
    HeaderFilterStrategy headerFilterStrategy = getHeaderFilterStrategy();
    //add all sqs message attributes as camel message headers so that knowledge of 
    //the Sqs class MessageAttributeValue will not leak to the client
    for (Entry<String, MessageAttributeValue> entry : msg.getMessageAttributes().entrySet()) {
        String header = entry.getKey();
        Object value = translateValue(entry.getValue());
        if (!headerFilterStrategy.applyFilterToExternalHeaders(header, value, exchange)) {
            message.setHeader(header, value);
        }
    }
    return exchange;
}
 

import com.amazonaws.services.sqs.model.MessageAttributeValue; //导入依赖的package包/类
@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));
}
 

import com.amazonaws.services.sqs.model.MessageAttributeValue; //导入依赖的package包/类
@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);
}
 

import com.amazonaws.services.sqs.model.MessageAttributeValue; //导入依赖的package包/类
@Test
public void shouldSendMessageWithCorrectAttributes() {
    //GIVEN
    String body = "Sample text message";
    Map<String, MessageAttributeValue> attributes = new HashMap<>();
    attributes.put("attribute1", new MessageAttributeValue()
            .withDataType("String")
            .withStringValue("value1"));
    attributes.put("attribute2", new MessageAttributeValue()
            .withDataType("Number")
            .withStringValue("230.000000000000000001"));

    //WHEN
    sender.send(body, attributes);

    //THEN
    SendMessageRequest expected = new SendMessageRequest();
    expected.withQueueUrl(queueUrl)
            .withMessageBody(body)
            .withMessageAttributes(attributes);
    verify(sqs).sendMessage(expected);
}
 

import com.amazonaws.services.sqs.model.MessageAttributeValue; //导入依赖的package包/类
@Test
public void shouldSendObjectMessageWithCorrectAttributes() throws JsonProcessingException {
    //GIVEN
    DummyObject bodyObject = new DummyObject();
    Map<String, MessageAttributeValue> attributes = new HashMap<>();
    attributes.put("attribute1", new MessageAttributeValue()
            .withDataType("String")
            .withStringValue("value1"));
    attributes.put("attribute2", new MessageAttributeValue()
            .withDataType("Number")
            .withStringValue("230.000000000000000001"));

    //WHEN
    sender.send(bodyObject, attributes);

    //THEN
    SendMessageRequest expected = new SendMessageRequest();
    expected.withQueueUrl(queueUrl)
            .withMessageBody(objectMapper.writeValueAsString(bodyObject))
            .withMessageAttributes(attributes);
    verify(sqs).sendMessage(expected);
}
 

import com.amazonaws.services.sqs.model.MessageAttributeValue; //导入依赖的package包/类
private int getMsgAttributesSize(Map<String, MessageAttributeValue> msgAttributes) {
	int totalMsgAttributesSize = 0;
	for (Entry<String, MessageAttributeValue> entry : msgAttributes.entrySet()) {
		totalMsgAttributesSize += getStringSizeInBytes(entry.getKey());

		MessageAttributeValue entryVal = entry.getValue();
		if (entryVal.getDataType() != null) {
			totalMsgAttributesSize += getStringSizeInBytes(entryVal.getDataType());
		}

		String stringVal = entryVal.getStringValue();
		if (stringVal != null) {
			totalMsgAttributesSize += getStringSizeInBytes(entryVal.getStringValue());
		}

		ByteBuffer binaryVal = entryVal.getBinaryValue();
		if (binaryVal != null) {
			totalMsgAttributesSize += binaryVal.array().length;
		}
	}
	return totalMsgAttributesSize;
}
 

import com.amazonaws.services.sqs.model.MessageAttributeValue; //导入依赖的package包/类
/***
 * Categorize the messages into batches per queue
 * @param messages
 * @return messageBatches - belonging to one or more queues
 */
private Map<String, List<SendMessageBatchRequestEntry>> createBatchesForQueues(final List<Message> messages) {
	
	final Map<String, List<SendMessageBatchRequestEntry>> messageBatches = new HashMap<String, List<SendMessageBatchRequestEntry>>();

	for(Message message : messages){
		final Map<String, MessageAttributeValue> attributes = this.toMessageAttrs(message);
		
		final SendMessageBatchRequestEntry entry = new SendMessageBatchRequestEntry() 
		.withId(message.getId())
		.withMessageAttributes(attributes)
		.withMessageBody(message.getBody());

		if(!messageBatches.containsKey(message.getQueue())){
			messageBatches.put(message.getQueue(), new ArrayList<SendMessageBatchRequestEntry>());
		}
		messageBatches.get(message.getQueue()).add(entry);
	}
	return messageBatches;
}
 

import com.amazonaws.services.sqs.model.MessageAttributeValue; //导入依赖的package包/类
private SendMessageRequest prepareSendMessageRequest(Message<?> message) {
    SendMessageRequest sendMessageRequest = new SendMessageRequest(this.queueUrl, String.valueOf(message.getPayload()));

    if (message.getHeaders().containsKey(SqsMessageHeaders.SQS_GROUP_ID_HEADER)) {
        sendMessageRequest.setMessageGroupId(message.getHeaders().get(SqsMessageHeaders.SQS_GROUP_ID_HEADER, String.class));
    }

    if (message.getHeaders().containsKey(SqsMessageHeaders.SQS_DEDUPLICATION_ID_HEADER)) {
        sendMessageRequest.setMessageDeduplicationId(message.getHeaders().get(SqsMessageHeaders.SQS_DEDUPLICATION_ID_HEADER, String.class));
    }

    if (message.getHeaders().containsKey(SqsMessageHeaders.SQS_DELAY_HEADER)) {
        sendMessageRequest.setDelaySeconds(message.getHeaders().get(SqsMessageHeaders.SQS_DELAY_HEADER, Integer.class));
    }

    Map<String, MessageAttributeValue> messageAttributes = getMessageAttributes(message);
    if (!messageAttributes.isEmpty()) {
        sendMessageRequest.withMessageAttributes(messageAttributes);
    }

    return sendMessageRequest;
}
 

import com.amazonaws.services.sqs.model.MessageAttributeValue; //导入依赖的package包/类
private static Map<String, Object> getMessageAttributesAsMessageHeaders(com.amazonaws.services.sqs.model.Message message) {
    Map<String, Object> messageHeaders = new HashMap<>();
    for (Map.Entry<String, MessageAttributeValue> messageAttribute : message.getMessageAttributes().entrySet()) {
        if (MessageHeaders.CONTENT_TYPE.equals(messageAttribute.getKey())) {
            messageHeaders.put(MessageHeaders.CONTENT_TYPE, MimeType.valueOf(messageAttribute.getValue().getStringValue()));
        } else if (MessageHeaders.ID.equals(messageAttribute.getKey())) {
            messageHeaders.put(MessageHeaders.ID, UUID.fromString(messageAttribute.getValue().getStringValue()));
        } else if (MessageAttributeDataTypes.STRING.equals(messageAttribute.getValue().getDataType())) {
            messageHeaders.put(messageAttribute.getKey(), messageAttribute.getValue().getStringValue());
        } else if (messageAttribute.getValue().getDataType().startsWith(MessageAttributeDataTypes.NUMBER)) {
            Object numberValue = getNumberValue(messageAttribute.getValue());
            if (numberValue != null) {
                messageHeaders.put(messageAttribute.getKey(), numberValue);
            }
        } else if (MessageAttributeDataTypes.BINARY.equals(messageAttribute.getValue().getDataType())) {
            messageHeaders.put(messageAttribute.getKey(), messageAttribute.getValue().getBinaryValue());
        }
    }

    return messageHeaders;
}
 

import com.amazonaws.services.sqs.model.MessageAttributeValue; //导入依赖的package包/类
@Test
public void receiveMessage_withMimeTypeMessageAttribute_shouldCopyToHeaders() throws Exception {
    // Arrange
    AmazonSQSAsync amazonSqs = mock(AmazonSQSAsync.class);
    MimeType mimeType = new MimeType("test", "plain", Charset.forName("UTF-8"));
    when(amazonSqs.receiveMessage(new ReceiveMessageRequest("http://testQueue").
            withWaitTimeSeconds(0).
            withMaxNumberOfMessages(1).
            withAttributeNames(QueueMessageChannel.ATTRIBUTE_NAMES).
            withMessageAttributeNames("All"))).
            thenReturn(new ReceiveMessageResult().withMessages(new com.amazonaws.services.sqs.model.Message().withBody("Hello").
                    withMessageAttributes(Collections.singletonMap(MessageHeaders.CONTENT_TYPE,
                            new MessageAttributeValue().withDataType(MessageAttributeDataTypes.STRING).withStringValue(mimeType.toString())))));

    PollableChannel messageChannel = new QueueMessageChannel(amazonSqs, "http://testQueue");

    // Act
    Message<?> receivedMessage = messageChannel.receive();

    // Assert
    assertEquals(mimeType, receivedMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE));
}
 

import com.amazonaws.services.sqs.model.MessageAttributeValue; //导入依赖的package包/类
@Test
public void receiveMessage_withStringMessageHeader_shouldBeReceivedAsQueueMessageAttribute() throws Exception {
    // Arrange
    AmazonSQSAsync amazonSqs = mock(AmazonSQSAsync.class);
    String headerValue = "Header value";
    String headerName = "MyHeader";
    when(amazonSqs.receiveMessage(new ReceiveMessageRequest("http://testQueue").
            withWaitTimeSeconds(0).
            withMaxNumberOfMessages(1).
            withAttributeNames(QueueMessageChannel.ATTRIBUTE_NAMES).
            withMessageAttributeNames("All"))).
            thenReturn(new ReceiveMessageResult().withMessages(new com.amazonaws.services.sqs.model.Message().withBody("Hello").
                    withMessageAttributes(Collections.singletonMap(headerName,
                            new MessageAttributeValue().withDataType(MessageAttributeDataTypes.STRING).withStringValue(headerValue)))));

    PollableChannel messageChannel = new QueueMessageChannel(amazonSqs, "http://testQueue");

    // Act
    Message<?> receivedMessage = messageChannel.receive();

    // Assert
    assertEquals(headerValue, receivedMessage.getHeaders().get(headerName));
}
 

import com.amazonaws.services.sqs.model.MessageAttributeValue; //导入依赖的package包/类
@Test
public void receiveMessage_withIncompatibleNumericMessageHeader_shouldThrowAnException() throws Exception {
    // Arrange
    AmazonSQSAsync amazonSqs = mock(AmazonSQSAsync.class);
    this.expectedException.expect(IllegalArgumentException.class);
    this.expectedException.expectMessage("Cannot convert String [17] to target class [java.util.concurrent.atomic.AtomicInteger]");

    HashMap<String, MessageAttributeValue> messageAttributes = new HashMap<>();
    AtomicInteger atomicInteger = new AtomicInteger(17);
    messageAttributes.put("atomicInteger", new MessageAttributeValue().withDataType(MessageAttributeDataTypes.NUMBER + ".java.util.concurrent.atomic.AtomicInteger").withStringValue(String.valueOf(atomicInteger)));

    when(amazonSqs.receiveMessage(new ReceiveMessageRequest("http://testQueue").
            withWaitTimeSeconds(0).
            withMaxNumberOfMessages(1).
            withAttributeNames(QueueMessageChannel.ATTRIBUTE_NAMES).
            withMessageAttributeNames("All"))).
            thenReturn(new ReceiveMessageResult().withMessages(new com.amazonaws.services.sqs.model.Message().withBody("Hello").
                    withMessageAttributes(messageAttributes)));

    PollableChannel messageChannel = new QueueMessageChannel(amazonSqs, "http://testQueue");

    // Act
    messageChannel.receive();
}
 

import com.amazonaws.services.sqs.model.MessageAttributeValue; //导入依赖的package包/类
@Test
public void receiveMessage_withMissingNumericMessageHeaderTargetClass_shouldThrowAnException() throws Exception {
    // Arrange
    AmazonSQSAsync amazonSqs = mock(AmazonSQSAsync.class);
    this.expectedException.expect(MessagingException.class);
    this.expectedException.expectMessage("Message attribute with value '12' and data type 'Number.class.not.Found' could not be converted" +
            " into a Number because target class was not found.");

    HashMap<String, MessageAttributeValue> messageAttributes = new HashMap<>();
    messageAttributes.put("classNotFound", new MessageAttributeValue().withDataType(MessageAttributeDataTypes.NUMBER + ".class.not.Found").withStringValue("12"));

    when(amazonSqs.receiveMessage(new ReceiveMessageRequest("http://testQueue").
            withWaitTimeSeconds(0).
            withMaxNumberOfMessages(1).
            withAttributeNames(QueueMessageChannel.ATTRIBUTE_NAMES).
            withMessageAttributeNames("All"))).
            thenReturn(new ReceiveMessageResult().withMessages(new com.amazonaws.services.sqs.model.Message().withBody("Hello").
                    withMessageAttributes(messageAttributes)));

    PollableChannel messageChannel = new QueueMessageChannel(amazonSqs, "http://testQueue");

    // Act
    messageChannel.receive();
}
 

import com.amazonaws.services.sqs.model.MessageAttributeValue; //导入依赖的package包/类
@Test
public void receiveMessage_withBinaryMessageHeader_shouldBeReceivedAsByteBufferMessageAttribute() throws Exception {
    // Arrange
    AmazonSQSAsync amazonSqs = mock(AmazonSQSAsync.class);
    ByteBuffer headerValue = ByteBuffer.wrap("My binary data!".getBytes());
    String headerName = "MyHeader";
    when(amazonSqs.receiveMessage(new ReceiveMessageRequest("http://testQueue").
            withWaitTimeSeconds(0).
            withMaxNumberOfMessages(1).
            withAttributeNames(QueueMessageChannel.ATTRIBUTE_NAMES).
            withMessageAttributeNames("All"))).
            thenReturn(new ReceiveMessageResult().withMessages(new com.amazonaws.services.sqs.model.Message().withBody("Hello").
                    withMessageAttributes(Collections.singletonMap(headerName,
                            new MessageAttributeValue().withDataType(MessageAttributeDataTypes.BINARY).withBinaryValue(headerValue)))));

    PollableChannel messageChannel = new QueueMessageChannel(amazonSqs, "http://testQueue");

    // Act
    Message<?> receivedMessage = messageChannel.receive();

    // Assert
    assertEquals(headerValue, receivedMessage.getHeaders().get(headerName));
}
 

import com.amazonaws.services.sqs.model.MessageAttributeValue; //导入依赖的package包/类
@Test
public void receiveMessage_withIdOfTypeString_IdShouldBeConvertedToUuid() throws Exception {
    // Arrange
    AmazonSQSAsync amazonSqs = mock(AmazonSQSAsync.class);
    UUID uuid = UUID.randomUUID();
    when(amazonSqs.receiveMessage(new ReceiveMessageRequest("http://testQueue").
            withWaitTimeSeconds(0).
            withMaxNumberOfMessages(1).
            withAttributeNames(QueueMessageChannel.ATTRIBUTE_NAMES).
            withMessageAttributeNames("All"))).
            thenReturn(new ReceiveMessageResult().withMessages(new com.amazonaws.services.sqs.model.Message().withBody("Hello").
                    withMessageAttributes(Collections.singletonMap(MessageHeaders.ID,
                            new MessageAttributeValue().withDataType(MessageAttributeDataTypes.STRING).withStringValue(uuid.toString())))));

    PollableChannel messageChannel = new QueueMessageChannel(amazonSqs, "http://testQueue");

    // Act
    Message<?> receivedMessage = messageChannel.receive();

    // Assert
    Object idMessageHeader = receivedMessage.getHeaders().get(MessageHeaders.ID);
    assertTrue(UUID.class.isInstance(idMessageHeader));
    assertEquals(uuid, idMessageHeader);
}
 

import com.amazonaws.services.sqs.model.MessageAttributeValue; //导入依赖的package包/类
/**
 * Adds the reply-to queue name and url attributes during send as part of the send message
 * request, if necessary
 */
private void addReplyToQueueReservedAttributes(Map<String, MessageAttributeValue> messageAttributes,
                                               SQSMessage message) throws JMSException {

    Destination replyTo = message.getJMSReplyTo();
    if (replyTo instanceof SQSQueueDestination) {
        SQSQueueDestination replyToQueue = (SQSQueueDestination)replyTo;

        /**
         * This will override the existing attributes if exists. Everything that
         * has prefix JMS_ is reserved for JMS Provider, but if the user sets that
         * attribute, it will be overwritten.
         */
        addStringAttribute(messageAttributes, SQSMessage.JMS_SQS_REPLY_TO_QUEUE_NAME, replyToQueue.getQueueName());
        addStringAttribute(messageAttributes, SQSMessage.JMS_SQS_REPLY_TO_QUEUE_URL, replyToQueue.getQueueUrl());
    }
}
 

import com.amazonaws.services.sqs.model.MessageAttributeValue; //导入依赖的package包/类
private Object translateValue(MessageAttributeValue mav) {
    Object result = null;
    if (mav.getStringValue() != null) {
        result = mav.getStringValue();
    } else if (mav.getBinaryValue() != null) {
        result = mav.getBinaryValue();
    }
    return result;
}
 

import com.amazonaws.services.sqs.model.MessageAttributeValue; //导入依赖的package包/类
@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);
    }
}
 

import com.amazonaws.services.sqs.model.MessageAttributeValue; //导入依赖的package包/类
public void send(Object object, Map<String, MessageAttributeValue> attributes) {
    final String json;
    try {
        json = objectMapper.writeValueAsString(object);
        send(json, attributes);
    } catch (JsonProcessingException e) {
        LOGGER.error("Could not send message to SQS, cause is " + e.getMessage(), e);
    }
}
 

import com.amazonaws.services.sqs.model.MessageAttributeValue; //导入依赖的package包/类
private SendMessageBatchRequestEntry storeMessageInS3(SendMessageBatchRequestEntry batchEntry) {

		checkMessageAttributes(batchEntry.getMessageAttributes());

		String s3Key = UUID.randomUUID().toString();

		// Read the content of the message from message body
		String messageContentStr = batchEntry.getMessageBody();

		Long messageContentSize = getStringSizeInBytes(messageContentStr);

		// Add a new message attribute as a flag
		MessageAttributeValue messageAttributeValue = new MessageAttributeValue();
		messageAttributeValue.setDataType("Number");
		messageAttributeValue.setStringValue(messageContentSize.toString());
		batchEntry.addMessageAttributesEntry(SQSExtendedClientConstants.RESERVED_ATTRIBUTE_NAME, messageAttributeValue);

		// Store the message content in S3.
		storeTextInS3(s3Key, messageContentStr, messageContentSize);

		LOG.info("S3 object created, Bucket name: " + clientConfiguration.getS3BucketName() + ", Object key: " + s3Key
				+ ".");

		// Convert S3 pointer (bucket name, key, etc) to JSON string
		MessageS3Pointer s3Pointer = new MessageS3Pointer(clientConfiguration.getS3BucketName(), s3Key);
		String s3PointerStr = getJSONFromS3Pointer(s3Pointer);

		// Storing S3 pointer in the message body.
		batchEntry.setMessageBody(s3PointerStr);

		return batchEntry;
	}