@Test void sendMessage_withNumericMessageHeaders_shouldBeSentAsTopicMessageAttributes() throws Exception { AmazonSNS amazonSns = mock(AmazonSNS.class); ArgumentCaptor<PublishRequest> publishRequestArgumentCaptor = ArgumentCaptor .forClass(PublishRequest.class); when(amazonSns.publish(publishRequestArgumentCaptor.capture())) .thenReturn(new PublishResult()); double doubleValue = 1234.56; long longValue = 1234L; int integerValue = 1234; byte byteValue = 2; short shortValue = 12; float floatValue = 1234.56f; BigInteger bigIntegerValue = new BigInteger("616416546156"); BigDecimal bigDecimalValue = new BigDecimal("7834938"); Message<String> message = MessageBuilder.withPayload("Hello") .setHeader("double", doubleValue).setHeader("long", longValue) .setHeader("integer", integerValue).setHeader("byte", byteValue) .setHeader("short", shortValue).setHeader("float", floatValue) .setHeader("bigInteger", bigIntegerValue) .setHeader("bigDecimal", bigDecimalValue).build(); MessageChannel messageChannel = new TopicMessageChannel(amazonSns, "topicArn"); boolean sent = messageChannel.send(message); assertThat(sent).isTrue(); Map<String, MessageAttributeValue> messageAttributes = publishRequestArgumentCaptor .getValue().getMessageAttributes(); assertThat(messageAttributes.get("double").getDataType()) .isEqualTo(MessageAttributeDataTypes.NUMBER + ".java.lang.Double"); assertThat(messageAttributes.get("double").getStringValue()) .isEqualTo(String.valueOf(doubleValue)); assertThat(messageAttributes.get("long").getDataType()) .isEqualTo(MessageAttributeDataTypes.NUMBER + ".java.lang.Long"); assertThat(messageAttributes.get("long").getStringValue()) .isEqualTo(String.valueOf(longValue)); assertThat(messageAttributes.get("integer").getDataType()) .isEqualTo(MessageAttributeDataTypes.NUMBER + ".java.lang.Integer"); assertThat(messageAttributes.get("integer").getStringValue()) .isEqualTo(String.valueOf(integerValue)); assertThat(messageAttributes.get("byte").getDataType()) .isEqualTo(MessageAttributeDataTypes.NUMBER + ".java.lang.Byte"); assertThat(messageAttributes.get("byte").getStringValue()) .isEqualTo(String.valueOf(byteValue)); assertThat(messageAttributes.get("short").getDataType()) .isEqualTo(MessageAttributeDataTypes.NUMBER + ".java.lang.Short"); assertThat(messageAttributes.get("short").getStringValue()) .isEqualTo(String.valueOf(shortValue)); assertThat(messageAttributes.get("float").getDataType()) .isEqualTo(MessageAttributeDataTypes.NUMBER + ".java.lang.Float"); assertThat(messageAttributes.get("float").getStringValue()) .isEqualTo(String.valueOf(floatValue)); assertThat(messageAttributes.get("bigInteger").getDataType()) .isEqualTo(MessageAttributeDataTypes.NUMBER + ".java.math.BigInteger"); assertThat(messageAttributes.get("bigInteger").getStringValue()) .isEqualTo(String.valueOf(bigIntegerValue)); assertThat(messageAttributes.get("bigDecimal").getDataType()) .isEqualTo(MessageAttributeDataTypes.NUMBER + ".java.math.BigDecimal"); assertThat(messageAttributes.get("bigDecimal").getStringValue()) .isEqualTo(String.valueOf(bigDecimalValue)); }
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 (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 SNS.", messageHeaderName, messageHeaderValue != null ? messageHeaderValue.getClass().getName() : "")); } } return messageAttributes; }
TopicMessageChannel extends AbstractMessageChannel { 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 (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 SNS.", messageHeaderName, messageHeaderValue != null ? messageHeaderValue.getClass().getName() : "")); } } return messageAttributes; } }
TopicMessageChannel extends AbstractMessageChannel { 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 (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 SNS.", messageHeaderName, messageHeaderValue != null ? messageHeaderValue.getClass().getName() : "")); } } return messageAttributes; } TopicMessageChannel(AmazonSNS amazonSns, String topicArn); }
TopicMessageChannel extends AbstractMessageChannel { 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 (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 SNS.", messageHeaderName, messageHeaderValue != null ? messageHeaderValue.getClass().getName() : "")); } } return messageAttributes; } TopicMessageChannel(AmazonSNS amazonSns, String topicArn);  }
TopicMessageChannel extends AbstractMessageChannel { 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 (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 SNS.", messageHeaderName, messageHeaderValue != null ? messageHeaderValue.getClass().getName() : "")); } } return messageAttributes; } TopicMessageChannel(AmazonSNS amazonSns, String topicArn);  static final String NOTIFICATION_SUBJECT_HEADER; }