@Test void sendMessage_withBinaryMessageHeader_shouldBeSentAsBinaryMessageAttribute() throws Exception { AmazonSQSAsync amazonSqs = mock(AmazonSQSAsync.class); QueueMessageChannel messageChannel = new QueueMessageChannel(amazonSqs, "http: 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()); boolean sent = messageChannel.send(message); assertThat(sent).isTrue(); assertThat(sendMessageRequestArgumentCaptor.getValue().getMessageAttributes() .get(headerName).getBinaryValue()).isEqualTo(headerValue); assertThat(sendMessageRequestArgumentCaptor.getValue().getMessageAttributes() .get(headerName).getDataType()) .isEqualTo(MessageAttributeDataTypes.BINARY); }
private Map<String, MessageAttributeValue> getMessageAttributes(Message<?> message) { HashMap<String, MessageAttributeValue> messageAttributes = new HashMap<>(); for (Map.Entry<String, Object> messageHeader : message.getHeaders().entrySet()) { String messageHeaderName = messageHeader.getKey(); Object messageHeaderValue = messageHeader.getValue(); if (isSkipHeader(messageHeaderName)) { continue; } if (MessageHeaders.CONTENT_TYPE.equals(messageHeaderName) && messageHeaderValue != null) { messageAttributes.put(messageHeaderName, getContentTypeMessageAttribute(messageHeaderValue)); } else if (MessageHeaders.ID.equals(messageHeaderName) && messageHeaderValue != null) { messageAttributes.put(messageHeaderName, getStringMessageAttribute(messageHeaderValue.toString())); } else if (messageHeaderValue instanceof String) { messageAttributes.put(messageHeaderName, getStringMessageAttribute((String) messageHeaderValue)); } else if (messageHeaderValue instanceof Number) { messageAttributes.put(messageHeaderName, getNumberMessageAttribute(messageHeaderValue)); } else if (messageHeaderValue instanceof ByteBuffer) { messageAttributes.put(messageHeaderName, getBinaryMessageAttribute((ByteBuffer) messageHeaderValue)); } else { this.logger.warn(String.format( "Message header with name '%s' and type '%s' cannot be sent as" + " message attribute because it is not supported by SQS.", messageHeaderName, messageHeaderValue != null ? messageHeaderValue.getClass().getName() : "")); } } return messageAttributes; }
QueueMessageChannel extends AbstractMessageChannel implements PollableChannel { private Map<String, MessageAttributeValue> getMessageAttributes(Message<?> message) { HashMap<String, MessageAttributeValue> messageAttributes = new HashMap<>(); for (Map.Entry<String, Object> messageHeader : message.getHeaders().entrySet()) { String messageHeaderName = messageHeader.getKey(); Object messageHeaderValue = messageHeader.getValue(); if (isSkipHeader(messageHeaderName)) { continue; } if (MessageHeaders.CONTENT_TYPE.equals(messageHeaderName) && messageHeaderValue != null) { messageAttributes.put(messageHeaderName, getContentTypeMessageAttribute(messageHeaderValue)); } else if (MessageHeaders.ID.equals(messageHeaderName) && messageHeaderValue != null) { messageAttributes.put(messageHeaderName, getStringMessageAttribute(messageHeaderValue.toString())); } else if (messageHeaderValue instanceof String) { messageAttributes.put(messageHeaderName, getStringMessageAttribute((String) messageHeaderValue)); } else if (messageHeaderValue instanceof Number) { messageAttributes.put(messageHeaderName, getNumberMessageAttribute(messageHeaderValue)); } else if (messageHeaderValue instanceof ByteBuffer) { messageAttributes.put(messageHeaderName, getBinaryMessageAttribute((ByteBuffer) messageHeaderValue)); } else { this.logger.warn(String.format( "Message header with name '%s' and type '%s' cannot be sent as" + " message attribute because it is not supported by SQS.", messageHeaderName, messageHeaderValue != null ? messageHeaderValue.getClass().getName() : "")); } } return messageAttributes; } }
QueueMessageChannel extends AbstractMessageChannel implements PollableChannel { private Map<String, MessageAttributeValue> getMessageAttributes(Message<?> message) { HashMap<String, MessageAttributeValue> messageAttributes = new HashMap<>(); for (Map.Entry<String, Object> messageHeader : message.getHeaders().entrySet()) { String messageHeaderName = messageHeader.getKey(); Object messageHeaderValue = messageHeader.getValue(); if (isSkipHeader(messageHeaderName)) { continue; } if (MessageHeaders.CONTENT_TYPE.equals(messageHeaderName) && messageHeaderValue != null) { messageAttributes.put(messageHeaderName, getContentTypeMessageAttribute(messageHeaderValue)); } else if (MessageHeaders.ID.equals(messageHeaderName) && messageHeaderValue != null) { messageAttributes.put(messageHeaderName, getStringMessageAttribute(messageHeaderValue.toString())); } else if (messageHeaderValue instanceof String) { messageAttributes.put(messageHeaderName, getStringMessageAttribute((String) messageHeaderValue)); } else if (messageHeaderValue instanceof Number) { messageAttributes.put(messageHeaderName, getNumberMessageAttribute(messageHeaderValue)); } else if (messageHeaderValue instanceof ByteBuffer) { messageAttributes.put(messageHeaderName, getBinaryMessageAttribute((ByteBuffer) messageHeaderValue)); } else { this.logger.warn(String.format( "Message header with name '%s' and type '%s' cannot be sent as" + " message attribute because it is not supported by SQS.", messageHeaderName, messageHeaderValue != null ? messageHeaderValue.getClass().getName() : "")); } } return messageAttributes; } QueueMessageChannel(AmazonSQSAsync amazonSqs, String queueUrl); }
QueueMessageChannel extends AbstractMessageChannel implements PollableChannel { private Map<String, MessageAttributeValue> getMessageAttributes(Message<?> message) { HashMap<String, MessageAttributeValue> messageAttributes = new HashMap<>(); for (Map.Entry<String, Object> messageHeader : message.getHeaders().entrySet()) { String messageHeaderName = messageHeader.getKey(); Object messageHeaderValue = messageHeader.getValue(); if (isSkipHeader(messageHeaderName)) { continue; } if (MessageHeaders.CONTENT_TYPE.equals(messageHeaderName) && messageHeaderValue != null) { messageAttributes.put(messageHeaderName, getContentTypeMessageAttribute(messageHeaderValue)); } else if (MessageHeaders.ID.equals(messageHeaderName) && messageHeaderValue != null) { messageAttributes.put(messageHeaderName, getStringMessageAttribute(messageHeaderValue.toString())); } else if (messageHeaderValue instanceof String) { messageAttributes.put(messageHeaderName, getStringMessageAttribute((String) messageHeaderValue)); } else if (messageHeaderValue instanceof Number) { messageAttributes.put(messageHeaderName, getNumberMessageAttribute(messageHeaderValue)); } else if (messageHeaderValue instanceof ByteBuffer) { messageAttributes.put(messageHeaderName, getBinaryMessageAttribute((ByteBuffer) messageHeaderValue)); } else { this.logger.warn(String.format( "Message header with name '%s' and type '%s' cannot be sent as" + " message attribute because it is not supported by SQS.", messageHeaderName, messageHeaderValue != null ? messageHeaderValue.getClass().getName() : "")); } } return messageAttributes; } QueueMessageChannel(AmazonSQSAsync amazonSqs, String queueUrl); @Override Message<String> receive(); @Override Message<String> receive(long timeout); }
QueueMessageChannel extends AbstractMessageChannel implements PollableChannel { private Map<String, MessageAttributeValue> getMessageAttributes(Message<?> message) { HashMap<String, MessageAttributeValue> messageAttributes = new HashMap<>(); for (Map.Entry<String, Object> messageHeader : message.getHeaders().entrySet()) { String messageHeaderName = messageHeader.getKey(); Object messageHeaderValue = messageHeader.getValue(); if (isSkipHeader(messageHeaderName)) { continue; } if (MessageHeaders.CONTENT_TYPE.equals(messageHeaderName) && messageHeaderValue != null) { messageAttributes.put(messageHeaderName, getContentTypeMessageAttribute(messageHeaderValue)); } else if (MessageHeaders.ID.equals(messageHeaderName) && messageHeaderValue != null) { messageAttributes.put(messageHeaderName, getStringMessageAttribute(messageHeaderValue.toString())); } else if (messageHeaderValue instanceof String) { messageAttributes.put(messageHeaderName, getStringMessageAttribute((String) messageHeaderValue)); } else if (messageHeaderValue instanceof Number) { messageAttributes.put(messageHeaderName, getNumberMessageAttribute(messageHeaderValue)); } else if (messageHeaderValue instanceof ByteBuffer) { messageAttributes.put(messageHeaderName, getBinaryMessageAttribute((ByteBuffer) messageHeaderValue)); } else { this.logger.warn(String.format( "Message header with name '%s' and type '%s' cannot be sent as" + " message attribute because it is not supported by SQS.", messageHeaderName, messageHeaderValue != null ? messageHeaderValue.getClass().getName() : "")); } } return messageAttributes; } QueueMessageChannel(AmazonSQSAsync amazonSqs, String queueUrl); @Override Message<String> receive(); @Override Message<String> receive(long timeout);  }