Java 类com.amazonaws.services.sns.model.PublishResult 实例源码

@Test
public void testPublish_shouldPublishTheMessage() {
  CreateTopicResult topicResult = sns.createTopic(new CreateTopicRequest().withName("important-topic"));

  PublishRequest publishReq = new PublishRequest()
      .withMessage("{\"state\":\"liquid\",\"color\":\"black\",\"waking-up\":true}")
      .withMessageStructure("json")
      .withPhoneNumber("00121212")
      .withSubject("eeffoc")
      .withTopicArn(topicResult.getTopicArn());

  PublishResult publishResult = sns.publish(publishReq);
  assertNotNull("verify message id",publishResult.getMessageId());

  Topic topic = getSnsTopics().get(topicResult.getTopicArn());
  assertEquals("verify correct message count", 1, topic.getMessages().size());
  assertEquals("verify message subject","eeffoc",topic.getMessages().get(0).getSubject());
  assertEquals("verify message body","{\"state\":\"liquid\",\"color\":\"black\",\"waking-up\":true}",topic.getMessages().get(0).getBody());
  assertEquals("verify message structure","json",topic.getMessages().get(0).getStructure());
}

public void process(Exchange exchange) throws Exception {
    PublishRequest request = new PublishRequest();

    request.setTopicArn(getConfiguration().getTopicArn());
    request.setSubject(determineSubject(exchange));
    request.setMessageStructure(determineMessageStructure(exchange));
    request.setMessage(exchange.getIn().getBody(String.class));

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

    PublishResult result = getEndpoint().getSNSClient().publish(request);

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

    Message message = getMessageForResponse(exchange);
    message.setHeader(SnsConstants.MESSAGE_ID, result.getMessageId());
}

@Override
public void execute(PostDownloadExecution execution) {

    AWSCredentials credentials = new BasicAWSCredentials(accessKey, secretAccessKey);

    AmazonSNS sns = snsClientBuilder.withRegion(region)
            .withCredentials(new AWSStaticCredentialsProvider(credentials)).build();

    LOGGER.debug("SNS: Topic Arn               : {}", arn);
    LOGGER.debug("SNS: Topic Region            : {}", region);
    LOGGER.debug("SNS: Topic Access Key        : {}", accessKey);
    LOGGER.debug("SNS: Topic Secret Access Key : {}", secretAccessKey);

    PublishRequest request = new PublishRequest();
    request.setTopicArn(arn);
    request.setMessageStructure("json");
    request.setMessage(formatJsonMessage(execution.fileName));
    request.setSubject("A new file has been downloaded");

    LOGGER.info("Publishing message to SNS");
    PublishResult result = sns.publish(request);
    LOGGER.info("Publish successful!");
    LOGGER.debug("{}", result.getMessageId());
}

@Override
public PublishResult publish(AwsParamsDto awsParamsDto, String topicArn, 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 snsOperations.publish(topicArn, messageText, messageAttributes, awsClientFactory.getAmazonSNSClient(awsParamsDto));
}

public PublishResult sendPushMessage(AmazonSNS snsClient, PushPlatform platform, String targetArn, Map<String, Object> customAppMessageMap, Map<String, MessageAttributeValue> messageAttributes) throws IOException {
    Map<String, Object> appMessageMap = new HashMap<String, Object>();

    if (platform == PushPlatform.GCM) {
        appMessageMap.put("collapse_key", "single");
        appMessageMap.put("delay_while_idle", true);
        appMessageMap.put("time_to_live", 30);
        appMessageMap.put("dry_run", false);
    }

    if (customAppMessageMap != null) {
        appMessageMap.putAll(customAppMessageMap);
    }

    String message = ServerUtil.asJson(appMessageMap);

    return sendPushMessage(snsClient, platform, targetArn, message, messageAttributes);
}

public PublishResult sendPushMessage(AmazonSNS snsClient, PushPlatform platform, String targetArn, String message,
        Map<String, MessageAttributeValue> messageAttributes) throws IOException {
    Map<String, String> messageMap = new HashMap<String, String>();
    messageMap.put(platform.name(), message);
    message = ServerUtil.asJson(messageMap);

    PublishRequest publishRequest = new PublishRequest();
    publishRequest.setMessageStructure("json");

    if (messageAttributes != null) {
        publishRequest.setMessageAttributes(messageAttributes);
    }

    publishRequest.setTargetArn(targetArn);
    publishRequest.setMessage(message);

    PublishResult publishResult = snsClient.publish(publishRequest);

    return publishResult;
}

@Test
public void sendMessage_withStringMessageHeader_shouldBeSentAsTopicMessageAttribute() throws Exception {
    // Arrange
    AmazonSNS amazonSns = mock(AmazonSNS.class);
    ArgumentCaptor<PublishRequest> publishRequestArgumentCaptor = ArgumentCaptor.forClass(PublishRequest.class);
    when(amazonSns.publish(publishRequestArgumentCaptor.capture())).thenReturn(new PublishResult());

    String headerValue = "Header value";
    String headerName = "MyHeader";
    Message<String> message = MessageBuilder.withPayload("Hello").setHeader(headerName, headerValue).build();
    MessageChannel messageChannel = new TopicMessageChannel(amazonSns, "topicArn");

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

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

@Test
public void sendMessage_withBinaryMessageHeader_shouldBeSentAsBinaryMessageAttribute() throws Exception {
    // Arrange
    AmazonSNS amazonSns = mock(AmazonSNS.class);
    ArgumentCaptor<PublishRequest> publishRequestArgumentCaptor = ArgumentCaptor.forClass(PublishRequest.class);
    when(amazonSns.publish(publishRequestArgumentCaptor.capture())).thenReturn(new PublishResult());

    ByteBuffer headerValue = ByteBuffer.wrap("My binary data!".getBytes());
    String headerName = "MyHeader";
    Message<String> message = MessageBuilder.withPayload("Hello").setHeader(headerName, headerValue).build();
    MessageChannel messageChannel = new TopicMessageChannel(amazonSns, "topicArn");

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

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

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

    ArgumentCaptor<PublishRequest> sendMessageRequestArgumentCaptor = ArgumentCaptor.forClass(PublishRequest.class);
    when(amazonSns.publish(sendMessageRequestArgumentCaptor.capture())).thenReturn(new PublishResult());

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

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

@Override
public PublishResult publish(PublishRequest publishRequest) throws AmazonClientException {
    String topicArn = publishRequest.getTopicArn();
    if (!_subscriptionsForTopic.containsKey(topicArn)) {
        throw new NotFoundException("no such topic " + topicArn);
    }
    List<Subscription> topicSubscriptions = FluentIterable.
            from(_subscriptionsForTopic.get(topicArn)).
            transform(Functions.forMap(_subscriptionsByArn)).
            toList();
    for (Subscription subscription : topicSubscriptions) {
        String queueName = getLast(subscription.getEndpoint().split(":"));
        String queueUrl = _sqsClient.
                getQueueUrl(new GetQueueUrlRequest().withQueueName(queueName)).
                getQueueUrl();
        _sqsClient.sendMessage(new SendMessageRequest().
                withQueueUrl(queueUrl).
                withMessageBody(publishRequest.getMessage()));
    }
    return new PublishResult();
}

public String sendNotification(CFNAssistNotification notification) throws CfnAssistException {
    String topicArn = getTopicARN();
    if (topicArn.isEmpty()) {
        logger.info("Will not send notification as sns topic not found, topic is: " + TOPIC_NAME);
        return "";
    }

    ObjectMapper objectMapper = new ObjectMapper();
    try {
        logger.info("Send notification: " + notification);
        String json = objectMapper.writeValueAsString(notification);
        PublishResult result = snsClient.publish(topicArn, json);
        logger.info(String.format("Send message on topic %s with id %s", TOPIC_NAME, result.getMessageId()));
        return result.getMessageId();
    }
    catch (JsonProcessingException jsonException) {
        throw new CfnAssistException("Unable to create notification JSON " + jsonException.toString());
    }
    catch (AuthorizationErrorException authException) {
        logger.error("Did not send SNS notification. You may need to update permissions for user via IAM. Exception was " 
                + authException);
        return "";
    }
}

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

    try {
        String serializedMessage = messageMarshaller.serialize(message);

        if (snsTestProxy == null) {
            PublishRequest request = new PublishRequest();
            PublishResult result = client.publish(request.withTopicArn(
                    topicArn).withMessage(serializedMessage));
            log.debug("Published message to topic: "
                    + result.getMessageId());
        } else {
            snsTestProxy.dispatchMessage(serializedMessage);
        }

    } catch (MessageMarshallerException e) {
        log.error(e.getMessage(), e);
        throw new MessagingException(e.getMessage(), e.getCause());
    }

    return message.getPayload();
}

private void demoNotification(Platform platform, String principal,
        String credential, String platformToken, String applicationName){
    // Create Platform Application. This corresponds to an app on a platform.
    CreatePlatformApplicationResult platformApplicationResult = createPlatformApplication(
            applicationName, platform, principal, credential);
    System.out.println(platformApplicationResult);

    // The Platform Application Arn can be used to uniquely identify the Platform Application.
    String platformApplicationArn = platformApplicationResult.getPlatformApplicationArn();

    // Create an Endpoint. This corresponds to an app on a device.
    CreatePlatformEndpointResult platformEndpointResult = createPlatformEndpoint(
            "CustomData - Useful to store endpoint specific data", platformToken, platformApplicationArn);
    System.out.println(platformEndpointResult);

    // Publish a push notification to an Endpoint.
    PublishResult publishResult = publish(platformEndpointResult.getEndpointArn(), platform);
    System.out.println("Published.  MessageId="+ publishResult.getMessageId());
    // Delete the Platform Application since we will no longer be using it.
    deletePlatformApplication(platformApplicationArn);
}

private PublishResult publish(String endpointArn, Platform platform) {
    PublishRequest publishRequest = new PublishRequest();
    Map<String, String> messageMap = new HashMap<String, String>();
    String message;
    messageMap.put("default", defaultMessage);
    messageMap.put(platform.name(), getPlatformSampleMessage(platform));
    // For direct publish to mobile end points, topicArn is not relevant.
    publishRequest.setTargetArn(endpointArn);
    publishRequest.setMessageStructure("json");
    message = jsonify(messageMap);

    // Display the message that will be sent to the endpoint/
    System.out.println(message);

    publishRequest.setMessage(message);
    return snsClient.publish(publishRequest);
}

public static Future<PublishResult> publish(String topic, String message) {
    String arn = String.format(ARN_TEMPLATE, properties.getProperty("aws.region"),
            properties.getProperty("aws.user.id"), topic);
    LOG.info("Sending SNS publish request to topic: {}", arn);
    PublishRequest publishRequest = new PublishRequest(arn, message);
    return SNS_CLIENT.publishAsync(publishRequest);
}

@Test
public void testDiscardOldest() throws Exception
{
    initialize("TestSNSAppender/testDiscardOldest.properties");

    // this is a dummy client: never actually run the writer thread, but
    // need to test the real writer
    MockSNSClient mockClient = new MockSNSClient("example", Arrays.asList("example"))
    {
        @Override
        protected PublishResult publish0(PublishRequest request)
        {
            throw new TestingException("this isn't going to work");
        }
    };

    appender.setThreadFactory(new NullThreadFactory());
    appender.setWriterFactory(mockClient.newWriterFactory());

    for (int ii = 0 ; ii < 20 ; ii++)
    {
        logger.debug("message " + ii);
    }

    List<LogMessage> messages = appender.getMessageQueue().toList();

    assertEquals("number of messages in queue", 10, messages.size());
    assertEquals("oldest message", "message 10\n", messages.get(0).getMessage());
    assertEquals("newest message", "message 19\n", messages.get(9).getMessage());
}

@Test
public void testDiscardNewest() throws Exception
{
    initialize("TestSNSAppender/testDiscardNewest.properties");

    // this is a dummy client: never actually run the writer thread, but
    // need to test the real writer
    MockSNSClient mockClient = new MockSNSClient("example", Arrays.asList("example"))
    {
        @Override
        protected PublishResult publish0(PublishRequest request)
        {
            throw new TestingException("this isn't going to work");
        }
    };

    appender.setThreadFactory(new NullThreadFactory());
    appender.setWriterFactory(mockClient.newWriterFactory());

    for (int ii = 0 ; ii < 20 ; ii++)
    {
        logger.debug("message " + ii);
    }

    List<LogMessage> messages = appender.getMessageQueue().toList();

    assertEquals("number of messages in queue", 10, messages.size());
    assertEquals("oldest message", "message 0\n", messages.get(0).getMessage());
    assertEquals("newest message", "message 9\n", messages.get(9).getMessage());
}

@Test
public void testDiscardNone() throws Exception
{
    initialize("TestSNSAppender/testDiscardNone.properties");

    // this is a dummy client: we never actually run the writer thread, but
    // need to test the real writer
    MockSNSClient mockClient = new MockSNSClient("example", Arrays.asList("example"))
    {
        @Override
        protected PublishResult publish0(PublishRequest request)
        {
            throw new TestingException("this isn't going to work");
        }
    };

    appender.setThreadFactory(new NullThreadFactory());
    appender.setWriterFactory(mockClient.newWriterFactory());

    for (int ii = 0 ; ii < 20 ; ii++)
    {
        logger.debug("message " + ii);
    }

    List<LogMessage> messages = appender.getMessageQueue().toList();

    assertEquals("number of messages in queue", 20, messages.size());
    assertEquals("oldest message", "message 0\n", messages.get(0).getMessage());
    assertEquals("newest message", "message 19\n", messages.get(19).getMessage());
}

/**
 * Sends using the sns publisher
 */
@Override
public boolean postMessage(NotificationMessage nMsg,
                           BasicSubscriber subscriber) {

    SNSSubscriber sub;
    if (subscriber instanceof SNSSubscriber) {
        sub = (SNSSubscriber) subscriber;
    } else {
        throw new ClassCastException("invalid subscriber " + subscriber.getClass().getName());
    }

    SNSMessage msg = buildSNSMessage(nMsg);
    AmazonSNS sns = new AmazonSNSClient(new BasicAWSCredentials(sub.getAwsAccessKey(), sub.getAwsSecretKey()));
    if (sub.getSnsEndpoint() != null) {
        sns.setEndpoint(sub.getSnsEndpoint());
    }

    CreateTopicRequest tRequest = new CreateTopicRequest();
    tRequest.setName(msg.getTopicName());
    CreateTopicResult result = sns.createTopic(tRequest);

    PublishRequest pr = new PublishRequest(result.getTopicArn(), msg.getTxtMessage()).withSubject(msg.getSubject());

    try {
        PublishResult pubresult = sns.publish(pr);
        logger.info("Published msg with id - " + pubresult.getMessageId());
    } catch (AmazonClientException ace) {
        logger.error(ace.getMessage());
        return false;
    }
    return true;
}

@Override
default void onSuccess(PublishRequest request, PublishResult result) {
    if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("SMS sent to: [{}]", request.getPhoneNumber());
        LOGGER.debug("SNS PublishResult messageId: [{}]", result.getMessageId());
    }
}

@Override
public SmsResponse sendSms(SmsRequest smsRequest) {
    try {
        PublishResult result = this.asyncSNS.publish(new PublishRequest()
                .withMessage(smsRequest.getMessage())
                .withPhoneNumber(smsRequest.getCountryCode() + smsRequest.getPhoneNumber())
                .withMessageAttributes(this.smsAttributes));
        return new SmsResponse(result.getMessageId(), result.getSdkHttpMetadata().getHttpStatusCode(),
                result.getSdkHttpMetadata().getHttpHeaders());
    } catch (Exception ex) {
        LOGGER.error("Exception while sending SMS!!", ex);
        throw new AwsException(ex.getMessage(), ex);
    }
}

public String handleRequest(Context context) throws IOException {
  LambdaLogger logger = context.getLogger();
  logger.log("handleRequest start.");
  ObjectMapper mapper = new ObjectMapper();
  JsonObject json = mapper.readValue(new URL(IP_RANGE_URL), JsonObject.class);

  AmazonS3Client s3 = getS3Client();
  Properties props = getProperties();
  s3.setEndpoint(props.getProperty("s3.endpoint"));

  GetObjectRequest request = new GetObjectRequest(props.getProperty(S3_BUCKETNAME_KEY), CHECKED_FILE_NAME);
  try {
    S3Object beforeObject = s3.getObject(request);
    InputStream is = beforeObject.getObjectContent();
    JsonObject beforeJson = mapper.readValue(is, JsonObject.class);
    Optional<String> diff = beforeJson.getDiff(json);
    if (diff.isPresent()) {
      AmazonSNSClient sns = getSNSClient();
      sns.setRegion(Region.getRegion(Regions.fromName(props.getProperty("sns.region"))));
      PublishRequest publishRequest = new PublishRequest(props.getProperty("sns.topic.arn"), diff.get(), SNS_SUBJECT);
      PublishResult result = sns.publish(publishRequest);
      logger.log("send sns message. messageId = " + result.getMessageId());
    }
  } catch (AmazonS3Exception e) {
    logger.log("before checked-ip-ranges.json does not exist.");
  }
  storeCheckedIpRange(json);
  logger.log("stored checked-ip-ranges.json");
  return "success";
}

@Test
public void testPublish()
{
    // Publish an SNS message without proxy.
    assertEquals(new PublishResult().withMessageId(MESSAGE_ID), snsDao.publish(new AwsParamsDto(), AWS_SNS_TOPIC_ARN, MESSAGE_TEXT, NO_MESSAGE_HEADERS));

    // Publish an SNS message using proxy settings.
    assertEquals(new PublishResult().withMessageId(MESSAGE_ID), snsDao
        .publish(new AwsParamsDto(NO_AWS_ACCESS_KEY, NO_AWS_SECRET_KEY, NO_SESSION_TOKEN, HTTP_PROXY_HOST, HTTP_PROXY_PORT), AWS_SNS_TOPIC_ARN,
            MESSAGE_TEXT, NO_MESSAGE_HEADERS));

    // Publish an SNS message with message headers.
    assertEquals(new PublishResult().withMessageId(MESSAGE_ID),
        snsDao.publish(new AwsParamsDto(), AWS_SNS_TOPIC_ARN, MESSAGE_TEXT, Collections.singletonList(new MessageHeader(KEY, VALUE))));
}

private PublishResult publish(String endpointArn, Platform platform,
        Map<Platform, Map<String, MessageAttributeValue>> attributesMap) {
    PublishRequest publishRequest = new PublishRequest();
    Map<String, MessageAttributeValue> notificationAttributes = getValidNotificationAttributes(attributesMap
            .get(platform));
    if (notificationAttributes != null && !notificationAttributes.isEmpty()) {
        publishRequest.setMessageAttributes(notificationAttributes);
    }
    publishRequest.setMessageStructure("json");
    // If the message attributes are not set in the requisite method,
    // notification is sent with default attributes
    String message = getPlatformSampleMessage(platform);
    Map<String, String> messageMap = new HashMap<String, String>();
    messageMap.put(platform.name(), message);
    message = SampleMessageGenerator.jsonify(messageMap);
    // For direct publish to mobile end points, topicArn is not relevant.
    publishRequest.setTargetArn(endpointArn);

    // Display the message that will be sent to the endpoint/
    System.out.println("{Message Body: " + message + "}");
    StringBuilder builder = new StringBuilder();
    builder.append("{Message Attributes: ");
    for (Map.Entry<String, MessageAttributeValue> entry : notificationAttributes
            .entrySet()) {
        builder.append("(\"" + entry.getKey() + "\": \""
                + entry.getValue().getStringValue() + "\"),");
    }
    builder.deleteCharAt(builder.length() - 1);
    builder.append("}");
    System.out.println(builder.toString());

    publishRequest.setMessage(message);
    return snsClient.publish(publishRequest);
}

public void demoNotification(Platform platform, String principal,
        String credential, String platformToken, String applicationName,
        Map<Platform, Map<String, MessageAttributeValue>> attrsMap) {
    // Create Platform Application. This corresponds to an app on a
    // platform.
    CreatePlatformApplicationResult platformApplicationResult = createPlatformApplication(
            applicationName, platform, principal, credential);
    System.out.println(platformApplicationResult);

    // The Platform Application Arn can be used to uniquely identify the
    // Platform Application.
    String platformApplicationArn = platformApplicationResult
            .getPlatformApplicationArn();

    // Create an Endpoint. This corresponds to an app on a device.
    CreatePlatformEndpointResult platformEndpointResult = createPlatformEndpoint(
            platform,
            "CustomData - Useful to store endpoint specific data",
            platformToken, platformApplicationArn);
    System.out.println(platformEndpointResult);

    // Publish a push notification to an Endpoint.
    PublishResult publishResult = publish(
            platformEndpointResult.getEndpointArn(), platform, attrsMap);
    System.out.println("Published! \n{MessageId="
            + publishResult.getMessageId() + "}");
    // Delete the Platform Application since we will no longer be using it.
    deletePlatformApplication(platformApplicationArn);
}

private PublishResult publish(String endpointArn, Platform platform,
        Map<Platform, Map<String, MessageAttributeValue>> attributesMap) {
    PublishRequest publishRequest = new PublishRequest();
    Map<String, MessageAttributeValue> notificationAttributes = getValidNotificationAttributes(attributesMap
            .get(platform));
    if (notificationAttributes != null && !notificationAttributes.isEmpty()) {
        //publishRequest.setMessageAttributes(notificationAttributes);

    }
    publishRequest.setMessageStructure("json");
    // If the message attributes are not set in the requisite method,
    // notification is sent with default attributes
    String message = getPlatformSampleMessage(platform);
    Map<String, String> messageMap = new HashMap<String, String>();
    messageMap.put(platform.name(), message);
    message = SampleMessageGenerator.jsonify(messageMap);
    // For direct publish to mobile end points, topicArn is not relevant.
    publishRequest.setTargetArn(endpointArn);

    // Display the message that will be sent to the endpoint/
    System.out.println("{Message Body: " + message + "}");
    StringBuilder builder = new StringBuilder();
    builder.append("{Message Attributes: ");
    for (Map.Entry<String, MessageAttributeValue> entry : notificationAttributes
            .entrySet()) {
        builder.append("(\"" + entry.getKey() + "\": \""
                + entry.getValue().getStringValue() + "\"),");
    }
    builder.deleteCharAt(builder.length() - 1);
    builder.append("}");
    System.out.println(builder.toString());

    publishRequest.setMessage(message);
    return snsClient.publish(publishRequest);
}

public void demoNotification(Platform platform, String principal,
        String credential, String platformToken, String applicationName,
        Map<Platform, Map<String, MessageAttributeValue>> attrsMap) {
    // Create Platform Application. This corresponds to an app on a
    // platform.
    CreatePlatformApplicationResult platformApplicationResult = createPlatformApplication(
            applicationName, platform, principal, credential);
    System.out.println(platformApplicationResult);

    // The Platform Application Arn can be used to uniquely identify the
    // Platform Application.
    String platformApplicationArn = platformApplicationResult
            .getPlatformApplicationArn();

    // Create an Endpoint. This corresponds to an app on a device.
    CreatePlatformEndpointResult platformEndpointResult = createPlatformEndpoint(
            platform,
            "CustomData - Useful to store endpoint specific data",
            platformToken, platformApplicationArn);
    System.out.println(platformEndpointResult);

    // Publish a push notification to an Endpoint.
    PublishResult publishResult = publish(
            platformEndpointResult.getEndpointArn(), platform, attrsMap);
    System.out.println("Published! \n{MessageId="
            + publishResult.getMessageId() + "}");
    // Delete the Platform Application since we will no longer be using it.
    deletePlatformApplication(platformApplicationArn);
}

@Override
protected void handle(final E event, final String encoded) throws Exception {
    PublishRequest request = new PublishRequest(topic, encoded, subject);
    String errorMessage = format("Appender '%s' failed to send logging event '%s' to SNS topic '%s'", getName(), event, topic);
    CountDownLatch latch = new CountDownLatch(isAsyncParent() ? 0 : 1);
    sns.publishAsync(request, new LoggingEventHandler<PublishRequest, PublishResult>(this, latch, errorMessage));
    AppenderExecutors.awaitLatch(this, latch, getMaxFlushTime());
}

@Override
public PublishResult publish(PublishRequest request,
        ResultCapture<PublishResult> extractor) {

    ActionResult result = resource.performAction("Publish", request,
            extractor);

    if (result == null) return null;
    return (PublishResult) result.getData();
}

@Override
public PublishResult publish(PublishRequest request,
        ResultCapture<PublishResult> extractor) {

    ActionResult result = resource.performAction("Publish", request,
            extractor);

    if (result == null) return null;
    return (PublishResult) result.getData();
}

@Override
public PublishResult publish(String message, ResultCapture<PublishResult>
        extractor) {

    PublishRequest request = new PublishRequest()
        .withMessage(message);
    return publish(request, extractor);
}

@Override
public PublishResult publish(String subject, String message,
        ResultCapture<PublishResult> extractor) {

    PublishRequest request = new PublishRequest()
        .withSubject(subject)
        .withMessage(message);
    return publish(request, extractor);
}

public void starterNotification(Platform platform, String principal,
        String credential, String platformToken, String applicationName,
        Map<Platform, Map<String, MessageAttributeValue>> attrsMap, String message) {
    // Create Platform Application. This corresponds to an app on a
    // platform.
    CreatePlatformApplicationResult platformApplicationResult = createPlatformApplication(
            applicationName, platform, principal, credential);
    System.out.println(platformApplicationResult);

    // The Platform Application Arn can be used to uniquely identify the
    // Platform Application.
    String platformApplicationArn = platformApplicationResult
            .getPlatformApplicationArn();

    // Create an Endpoint. This corresponds to an app on a device.
    CreatePlatformEndpointResult platformEndpointResult = createPlatformEndpoint(
            platform,
            "CustomData - Useful to store endpoint specific data",
            platformToken, platformApplicationArn);
    System.out.println(platformEndpointResult);

    // Publish a push notification to an Endpoint.
    PublishResult publishResult = publish(
            platformEndpointResult.getEndpointArn(), platform, attrsMap, message);
    System.out.println("Published! \n{MessageId="
            + publishResult.getMessageId() + "}");
    // Delete the Platform Application since we will no longer be using it.
    deletePlatformApplication(platformApplicationArn);
}

private <T extends Serializable> void sendMessageToAllRegionsSync(final T body) throws IOException {
    if ( sns == null ) {
        logger.error( "SNS client is null, perhaps it failed to initialize successfully" );
        return;
    }

    final String stringBody = toString( body );

    String topicArn = getWriteTopicArn();

    if ( logger.isTraceEnabled() ) {
        logger.trace( "Publishing Message...{} to arn: {}", stringBody, topicArn );
    }

    try {
        PublishResult publishResult = sns.publish(topicArn, toString(body));
        if ( logger.isTraceEnabled() ) {
            logger.trace( "Successfully published... messageID=[{}],  arn=[{}]", publishResult.getMessageId(),
                topicArn );
        }
    } catch (Exception e) {
        if (logger.isErrorEnabled()) {
            logger.error(FAILED_TO_SEND_MESSAGE,  stringBody, topicArn, e);
        }
        sendMessageToAllRegionsAsync(body);
    }



}

private <T extends Serializable> void sendMessageToAllRegionsAsync(final T body ) throws IOException {
    if ( snsAsync == null ) {
        logger.error( "SNS client is null, perhaps it failed to initialize successfully" );
        return;
    }

    final String stringBody = toString( body );

    final String topicArn = getWriteTopicArn();

    if ( logger.isTraceEnabled() ) {
        logger.trace( "Publishing Message...{} to arn: {}", stringBody, topicArn );
    }

    PublishRequest publishRequest = new PublishRequest( topicArn, stringBody );

    snsAsync.publishAsync( publishRequest, new AsyncHandler<PublishRequest, PublishResult>() {
        @Override
        public void onError( Exception e ) {
            logger.error( "Error publishing message... {}", e );
            logger.error(FAILED_TO_SEND_MESSAGE, stringBody, topicArn, e);
        }


        @Override
        public void onSuccess( PublishRequest request, PublishResult result ) {
            if ( logger.isTraceEnabled() ) {
                logger.trace( "Successfully published... messageID=[{}],  arn=[{}]", result.getMessageId(),
                    request.getTopicArn() );
            }
        }
    } );
}

private void publishNotification(
    final String arn,
    final SNSMessage<?> message,
    final QualifiedName name,
    final String errorMessage,
    final String counterKey,
    final boolean retryOnLongMessage
) {
    this.notificationMetric.recordTime(message, Metrics.TimerNotificationsBeforePublishDelay.getMetricName());
    try {
        final AsyncHandler<PublishRequest, PublishResult> handler =
            new AsyncHandler<PublishRequest, PublishResult>() {
                @Override
                public void onError(final Exception exception) {
                    if (retryOnLongMessage && (exception instanceof InvalidParameterException
                        || exception instanceof InvalidParameterValueException)) {
                        log.error("SNS Publish message exceeded the size threshold", exception);
                        notificationMetric.counterIncrement(
                            Metrics.CounterSNSNotificationPublishMessageSizeExceeded.getMetricName());
                        final SNSMessage<Void> voidMessage = new SNSMessage<>(message.getId(),
                            message.getTimestamp(), message.getRequestId(), message.getType(), message.getName(),
                            null);
                        publishNotification(arn, voidMessage, name, errorMessage, counterKey, false);
                    } else {
                        notificationMetric.handleException(name, errorMessage, counterKey, message, exception);
                    }
                }

                @Override
                public void onSuccess(final PublishRequest request, final PublishResult publishResult) {
                    log.info("Successfully published message to topic {} with id {}",
                        arn, publishResult.getMessageId());
                    log.debug("Successfully published message {} to topic {} with id {}",
                        message, arn, publishResult.getMessageId());
                    notificationMetric.counterIncrement(counterKey);
                    notificationMetric.recordTime(message,
                        Metrics.TimerNotificationsPublishDelay.getMetricName());
                }
            };
        client.publishAsync(arn, mapper.writeValueAsString(message), handler);
    } catch (final Exception e) {
        notificationMetric.handleException(name, errorMessage, counterKey, message, e);
    }
}

@Override
public PublishResult publish(PublishRequest publishRequest) throws AmazonServiceException, AmazonClientException {
    PublishResult publishResult = new PublishResult();
    publishResult.setMessageId("dcc8ce7a-7f18-4385-bedd-b97984b4363c");
    return publishResult;
}

/**
 * Publishes a notification message.
 *
 * @param notificationMessage the notification message
 */
protected void publishNotificationMessageImpl(NotificationMessage notificationMessage)
{
    try
    {
        // Send notification message to the specified destination.
        if (notificationMessage.getMessageType().equals(MessageTypeEntity.MessageEventTypes.SQS.name()))
        {
            // Publish the message using SQS.
            SendMessageResult sendMessageResult = sqsDao
                .sendMessage(awsHelper.getAwsParamsDto(), notificationMessage.getMessageDestination(), notificationMessage.getMessageText(),
                    notificationMessage.getMessageHeaders());

            // Log response and the message information.
            LOGGER.info("Published {} notification message. messageDestination=\"{}\" messageText={} messageHeaders={} sendMessageResult={}",
                notificationMessage.getMessageType(), notificationMessage.getMessageDestination(), notificationMessage.getMessageText(),
                jsonHelper.objectToJson(notificationMessage.getMessageHeaders()), jsonHelper.objectToJson(sendMessageResult));
        }
        else if (notificationMessage.getMessageType().equals(MessageTypeEntity.MessageEventTypes.SNS.name()))
        {
            // Publish the message using SNS.
            PublishResult publishResult = snsDao
                .publish(awsHelper.getAwsParamsDto(), notificationMessage.getMessageDestination(), notificationMessage.getMessageText(),
                    notificationMessage.getMessageHeaders());

            // Log response and the message information.
            LOGGER.info("Published {} notification message. messageDestination=\"{}\" messageText={} messageHeaders={} publishResult={}",
                notificationMessage.getMessageType(), notificationMessage.getMessageDestination(), notificationMessage.getMessageText(),
                jsonHelper.objectToJson(notificationMessage.getMessageHeaders()), jsonHelper.objectToJson(publishResult));
        }
        else
        {
            throw new IllegalStateException(String.format("Notification message type \"%s\" is not supported.", notificationMessage.getMessageType()));
        }
    }
    catch (RuntimeException e)
    {
        // Log an error message.
        LOGGER.error("Failed to publish {} notification message to \"{}\" destination. messageText={} messageHeaders={}",
            notificationMessage.getMessageType(), notificationMessage.getMessageDestination(), notificationMessage.getMessageText(),
            jsonHelper.objectToJson(notificationMessage.getMessageHeaders()));

        // Rethrow the exception.
        throw new IllegalStateException(e.getMessage(), e);
    }
}

@Override
public PublishResult publish(String topicArn, String messageText, Map<String, MessageAttributeValue> messageAttributes, AmazonSNS amazonSNS)
{
    return amazonSNS.publish(new PublishRequest().withTopicArn(topicArn).withMessage(messageText).withMessageAttributes(messageAttributes));
}

@Override
public PublishResult publish(String topicArn, String messageText, Map<String, MessageAttributeValue> messageAttributes, AmazonSNS amazonSNS)
{
    // Nothing to do in the normal case since our unit tests aren't reading messages once they have been published.
    return new PublishResult().withMessageId(AbstractDaoTest.MESSAGE_ID);
}