SnsNotificationSender implements NotificationSender { @Override public void send(DataHighwayNotification notification) { String topicArn = topicArnFactory.topicArn(notification); try { final String jsonMessage = objectWriter.writeValueAsString(notification); if (jsonMessage.getBytes(Charsets.UTF_8).length > SNS_MESSAGE_SIZE_LIMIT_BYTES) { log.error("Message length exceeds SNS limit ({} bytes).", SNS_MESSAGE_SIZE_LIMIT_BYTES); } log.debug("Attempting to send message to topic '{}': {}", topicArn, jsonMessage); String message = messageFactory.message(topicArn, notification); PublishRequest request = new PublishRequest(topicArn, jsonMessage, message); request.addMessageAttributesEntry(PROTOCOL_VERSION, attributeStringValue(notification.getProtocolVersion())); request.addMessageAttributesEntry(TYPE, attributeStringValue(notification.getType().name())); request.addMessageAttributesEntry(ROAD_NAME, attributeStringValue(notification.getRoadName())); try { sns.publish(request); } catch (AmazonClientException e) { log.error("Could not publish message '{}' to SNS topic '{}'.", message, topicArn, e); } } catch (JsonProcessingException e) { log.error("Could not serialize message '{}'.", notification, e); } } SnsNotificationSender(AmazonSNSAsync sns, TopicArnFactory topicArnFactory, MessageFactory messageFactory); @Override void send(DataHighwayNotification notification);  }
@Test public void typical() throws Exception { snsSender = new SnsNotificationSender(sns, n -> String.format("topicArn=%s", n.getRoadName()), (t, n) -> String.format("subject=%s", n.getRoadName())); HivePartitionCreated notification = HivePartitionCreated .builder() .databaseName(DATABASE_NAME) .locationUri(LOCATION_URI) .metastoreUris(HIVE_METASTORE_URIS) .partitionSpec(PARTITION_SPEC) .roadName(ROAD_NAME) .tableName(TABLE_NAME) .recordCount(1L) .build(); when(sns.publish(requestCaptor.capture())).thenReturn(null); snsSender.send(notification); PublishRequest request = requestCaptor.getValue(); assertThat(request.getTopicArn(), is("topicArn=" + ROAD_NAME)); assertThat(request.getSubject(), is("subject=" + ROAD_NAME)); @SuppressWarnings("unchecked") Map<String, Object> messageMap = new ObjectMapper().readValue(request.getMessage(), HashMap.class); assertThat(messageMap.keySet(), is(ImmutableSet.of("protocolVersion", "type", "roadName", "metastoreUris", "databaseName", "tableName", "partitionSpec", "locationUri", "recordCount"))); assertThat(messageMap, hasEntry("protocolVersion", "1.0")); assertThat(messageMap, hasEntry("type", RoadNotificationType.HIVE_PARTITION_CREATED.name())); assertThat(messageMap, hasEntry("roadName", ROAD_NAME)); assertThat(messageMap, hasEntry("metastoreUris", HIVE_METASTORE_URIS)); assertThat(messageMap, hasEntry("databaseName", DATABASE_NAME)); assertThat(messageMap, hasEntry("tableName", TABLE_NAME)); assertThat(messageMap, hasEntry("partitionSpec", PARTITION_SPEC)); assertThat(messageMap, hasEntry("locationUri", LOCATION_URI)); assertThat(messageMap, hasEntry("recordCount", 1)); Map<String, MessageAttributeValue> messageAttributes = request.getMessageAttributes(); assertThat(messageAttributes.get(SnsNotificationSender.PROTOCOL_VERSION).getStringValue(), is("1.0")); assertThat(messageAttributes.get(SnsNotificationSender.PROTOCOL_VERSION).getDataType(), is("String")); assertThat(messageAttributes.get(SnsNotificationSender.TYPE).getStringValue(), is("HIVE_PARTITION_CREATED")); assertThat(messageAttributes.get(SnsNotificationSender.TYPE).getDataType(), is("String")); assertThat(messageAttributes.get(SnsNotificationSender.ROAD_NAME).getStringValue(), is("ROAD_NAME")); assertThat(messageAttributes.get(SnsNotificationSender.ROAD_NAME).getDataType(), is("String")); }