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

@Override
public void run(Webhook webhook, String login, String email, String name, String subject, String content) {

    SNSWebhookApp webhookApp = (SNSWebhookApp) webhook.getWebhookApp();
    String topicArn = webhookApp.getTopicArn();
    String awsAccount = webhookApp.getAwsAccount();
    String awsSecret = webhookApp.getAwsSecret();
    String region = webhookApp.getRegion();
    AmazonSNSClient snsClient = new AmazonSNSClient(new BasicAWSCredentials(awsAccount, awsSecret));
    snsClient.setRegion(Region.getRegion(Regions.fromName(region)));

    try {
        PublishRequest publishReq = new PublishRequest()
                .withTopicArn(topicArn)
                .withMessage(getMessage(login, email, name, subject, content));
        snsClient.publish(publishReq);

    } catch (Exception e) {
        LOGGER.log(Level.SEVERE, "Cannot send notification to SNS service", e);
    } finally {
        LOGGER.log(Level.INFO, "Webhook runner terminated");
    }
}

@Test
public void start() {
  SnsListener listener = new SnsListener(client, config, clock);
  listener.circusTrainStartUp(new String[] {}, sourceCatalog, replicaCatalog);
  listener.tableReplicationStart(tableReplication, EVENT_ID);

  verify(client).publish(requestCaptor.capture());
  PublishRequest request = requestCaptor.getValue();
  assertThat(request.getSubject(), is(SUBJECT));
  assertThat(request.getTopicArn(), is("startArn"));
  assertThat(request.getMessage(),
      is("{\"protocolVersion\":\"1.0\",\"type\":\"START\",\"headers\":{\"pipeline-id\":\"0943879438\"},"
          + "\"startTime\":\"starttime\",\"eventId\":\"EVENT_ID\",\"sourceCatalog\":\"sourceCatalogName\","
          + "\"replicaCatalog\":\"replicaCatalogName\",\"sourceTable\":\"srcDb.srcTable\",\"replicaTable\":"
          + "\"replicaDb.replicaTable\"}"));
}

@Test
public void success() {
  SnsListener listener = new SnsListener(client, config, clock);
  listener.circusTrainStartUp(new String[] {}, sourceCatalog, replicaCatalog);
  listener.tableReplicationStart(tableReplication, EVENT_ID);
  listener.partitionsToAlter(PARTITIONS_0);
  listener.partitionsToCreate(PARTITIONS_1);
  listener.copierEnd(metrics);
  listener.tableReplicationSuccess(tableReplication, EVENT_ID);

  verify(client, times(2)).publish(requestCaptor.capture());
  PublishRequest request = requestCaptor.getAllValues().get(1);
  assertThat(request.getSubject(), is(SUBJECT));
  assertThat(request.getTopicArn(), is("successArn"));
  assertThat(request.getMessage(),
      is("{\"protocolVersion\":\"1.0\",\"type\":\"SUCCESS\",\"headers\":{\"pipeline-id\":\"0943879438\"},"
          + "\"startTime\":\"starttime\",\"endTime\":\"endtime\",\"eventId\":\"EVENT_ID\",\"sourceCatalog\""
          + ":\"sourceCatalogName\",\"replicaCatalog\":\"replicaCatalogName\",\"sourceTable\":"
          + "\"srcDb.srcTable\",\"replicaTable\":\"replicaDb.replicaTable\",\"modifiedPartitions\":"
          + "[[\"2014-01-01\",\"0\"],[\"2014-01-01\",\"1\"]],\"bytesReplicated\":40}"));
}

@Test
public void failure() {
  SnsListener listener = new SnsListener(client, config, clock);
  listener.circusTrainStartUp(new String[] {}, sourceCatalog, replicaCatalog);
  listener.tableReplicationStart(tableReplication, EVENT_ID);
  listener.partitionsToAlter(PARTITIONS_0);
  listener.partitionsToCreate(PARTITIONS_1);
  listener.copierEnd(metrics);
  listener.tableReplicationFailure(tableReplication, EVENT_ID, ERROR);

  verify(client, times(2)).publish(requestCaptor.capture());
  PublishRequest request = requestCaptor.getValue();
  assertThat(request.getSubject(), is(SUBJECT));
  assertThat(request.getTopicArn(), is("failArn"));
  assertThat(request.getMessage(), is("{\"protocolVersion\":\"1.0\",\"type\":\"FAILURE\",\"headers\":"
      + "{\"pipeline-id\":\"0943879438\"},\"startTime\":\"starttime\",\"endTime\":\"endtime\",\"eventId\":"
      + "\"EVENT_ID\",\"sourceCatalog\":\"sourceCatalogName\",\"replicaCatalog\":\"replicaCatalogName\","
      + "\"sourceTable\":\"srcDb.srcTable\",\"replicaTable\":\"replicaDb.replicaTable\",\"modifiedPartitions\":"
      + "[[\"2014-01-01\",\"0\"],[\"2014-01-01\",\"1\"]],\"bytesReplicated\":40,\"errorMessage\":\"error message\"}"));
}

@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());
}

@Override
public void process(PluginContext ctx, TenantId tenantId, RuleId ruleId, RuleToPluginMsg<?> msg) throws RuleException {
    if (msg instanceof SnsTopicActionMsg) {
        SnsTopicActionPayload payload = ((SnsTopicActionMsg) msg).getPayload();
        PublishRequest publishRequest = new PublishRequest()
                .withTopicArn(payload.getTopicArn())
                .withMessage(payload.getMsgBody());
        sns.publish(publishRequest);
        if (payload.isSync()) {
            ctx.reply(new ResponsePluginToRuleMsg(msg.getUid(), tenantId, ruleId,
                    BasicStatusCodeResponse.onSuccess(payload.getMsgType(), payload.getRequestId())));
        }
       return;
    }
    throw new RuleException("Unsupported message type " + msg.getClass().getName() + "!");

}

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());
}

public void notify(String topicId, String title, String message) {

        try {
            if (topicId == null)
                Utils.warn("Unable to send internal notification [%s][%s]: no SNS topic id.", //
                        title, message);
            else
                snsClient.publish(new PublishRequest()//
                        .withTopicArn(topicId)//
                        .withSubject(title)//
                        .withMessage(message));

        } catch (Throwable ignore) {
            ignore.printStackTrace();
        }
    }

@Test
public void shouldPublish_withSubjectAndMessage() {
    // Given
    final String subject = randomString();
    final String message = randomString();

    // When
    snsTopicResource.publish(subject, message);

    // Then
    final ArgumentCaptor<PublishRequest> captor = ArgumentCaptor.forClass(PublishRequest.class);
    verify(mockAmazonSnsClient).publish(captor.capture());
    final PublishRequest publishRequest = captor.getValue();
    assertEquals(topicArn, publishRequest.getTopicArn());
    assertEquals(subject, publishRequest.getSubject());
    assertEquals(message, publishRequest.getMessage());
    assertEquals(subject, publishRequest.getMessageAttributes().get("subject").getStringValue());
    assertEquals("String", publishRequest.getMessageAttributes().get("subject").getDataType());
}

@Test
public void shouldThrowException_onAmazonClientExceptionFromPublish() {
    // Given
    final String subject = randomString();
    final String message = randomString();
    when(mockAmazonSnsClient.publish(any(PublishRequest.class))).thenThrow(AmazonClientException.class);

    // When
    AmazonClientException thrownException = null;
    try {
        snsTopicResource.publish(subject, message);
    } catch (final AmazonClientException e) {
        thrownException = e;
    }

    // Then
    assertNotNull(thrownException);
}

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
@Ignore
public void testActions() {

    // setAttribtues
    String userData = UUID.randomUUID().toString();
    endpoint.setAttributes(new SetEndpointAttributesRequest()
            .addAttributesEntry("CustomUserData", userData)
            );
    refreshEndpoint();
    Assert.assertEquals(userData, endpoint.getAttributes().get("CustomUserData"));

    // publish
    endpoint.publish(new PublishRequest()
            .withSubject("subject")
            .withMessage("message")
            );
}

@Test
public void send_validTextMessage_usesTopicChannel() throws Exception {
    // Arrange
    AmazonSNS amazonSns = mock(AmazonSNS.class);
    NotificationMessagingTemplate notificationMessagingTemplate = new NotificationMessagingTemplate(amazonSns);
    String physicalTopicName = "arn:aws:sns:eu-west:123456789012:test";
    when(amazonSns.listTopics(new ListTopicsRequest(null))).thenReturn(new ListTopicsResult().withTopics(new Topic().withTopicArn(physicalTopicName)));
    notificationMessagingTemplate.setDefaultDestinationName(physicalTopicName);

    // Act
    notificationMessagingTemplate.send(MessageBuilder.withPayload("Message content").build());

    // Assert
    verify(amazonSns).publish(new PublishRequest(physicalTopicName,
            "Message content", null).withMessageAttributes(isNotNull()));
}

@Test
public void sendMessage_validTextMessageAndSubject_returnsTrue() throws Exception {
    // Arrange
    AmazonSNS amazonSns = mock(AmazonSNS.class);

    Message<String> stringMessage = MessageBuilder.withPayload("Message content").setHeader(TopicMessageChannel.NOTIFICATION_SUBJECT_HEADER, "Subject").build();
    MessageChannel messageChannel = new TopicMessageChannel(amazonSns, "topicArn");

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

    // Assert
    verify(amazonSns, only()).publish(new PublishRequest("topicArn",
            "Message content", "Subject").withMessageAttributes(isNotNull()));
    assertTrue(sent);
}

@Test
public void sendMessage_validTextMessageWithoutSubject_returnsTrue() throws Exception {
    // Arrange
    AmazonSNS amazonSns = mock(AmazonSNS.class);

    Message<String> stringMessage = MessageBuilder.withPayload("Message content").build();
    MessageChannel messageChannel = new TopicMessageChannel(amazonSns, "topicArn");

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

    // Assert
    verify(amazonSns, only()).publish(new PublishRequest("topicArn",
            "Message content", null).withMessageAttributes(isNotNull()));
    assertTrue(sent);
}

@Test
public void sendMessage_validTextMessageAndTimeout_timeoutIsIgnored() throws Exception {
    // Arrange
    AmazonSNS amazonSns = mock(AmazonSNS.class);

    Message<String> stringMessage = MessageBuilder.withPayload("Message content").build();
    MessageChannel messageChannel = new TopicMessageChannel(amazonSns, "topicArn");

    // Act
    boolean sent = messageChannel.send(stringMessage, 10);

    // Assert
    verify(amazonSns, only()).publish(new PublishRequest("topicArn",
            "Message content", null).withMessageAttributes(isNotNull()));
    assertTrue(sent);
}

@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());
}

/**
 * Posts message to queue.
 *
 * @param message Message to post.
 * @param eventName Message subject (type of message).
 * @throws MessagingException Failed to post message.
 */
@Override
public <M> void post(M message, String eventName) throws MessagingException {

    if (empty(eventName)) {
        throw new MessagingException("Cannot publish message with empty eventName!");
    }
    try {
        String jsonBody = messageSerializer.serialize(message);
        String encryptedBody = messageSerializer.encrypt(jsonBody);
        PublishRequest publishRequest = new PublishRequest(topicArn, encryptedBody, eventName);
        amazonSNS.publish(publishRequest);
    } catch (Exception e) {
        throw new MessagingException("Failed to publish message " + eventName, e);
    }
}

@Test
public void shouldUseMessageTypeAsSubjectWhenPostingToSNS() throws Exception {

    // Arrange
    TestMessage message = new TestMessage("ad99bb4f");
    MessagePublisher publisherService = PublisherService.nonEncryptedPublisherService(snsClient, "topicArn");

    // Act
    publisherService.post(message, TestMessage.class.getName());

    // Assert
    ArgumentCaptor<PublishRequest> argumentCaptor = ArgumentCaptor.forClass(PublishRequest.class);
    verify(snsClient).publish(argumentCaptor.capture());
    assertEquals("topicArn", argumentCaptor.getValue().getTopicArn());
    assertEquals(TestMessage.class.getName(), argumentCaptor.getValue().getSubject());
    assertEquals("{\"message\":\"ad99bb4f\"}", argumentCaptor.getValue().getMessage());
}

@Test
public void shouldUseSpecifiedEventNameAsSubjectWhenPostingToSNS() throws Exception {

    // Arrange
    TestMessage message = new TestMessage("ad99bb4f");
    MessagePublisher publisherService = PublisherService.nonEncryptedPublisherService(snsClient, "topicArn");

    // Act
    publisherService.post(message, "ForcedEventName");

    // Assert
    ArgumentCaptor<PublishRequest> argumentCaptor = ArgumentCaptor.forClass(PublishRequest.class);
    verify(snsClient).publish(argumentCaptor.capture());
    assertEquals("topicArn", argumentCaptor.getValue().getTopicArn());
    assertEquals("ForcedEventName", argumentCaptor.getValue().getSubject());
    assertEquals("{\"message\":\"ad99bb4f\"}", argumentCaptor.getValue().getMessage());
}

@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 void publishAndReceiveSeparateSQSClients() {
    final String queueName = someQueueName();
    final String queueUrl = someNewQueue(queueName);
    final String topicName = "publishAndReceiveSeparateSQSClients";
    final String message = "hi from " + topicName;

    AmazonSNS amazonSNS = new InMemorySNS(_amazonSQS1,
            new Subscription().
                    withTopicArn(makeTopicArn(topicName)).
                    withProtocol("sqs").
                    withSubscriptionArn(makeSomeSubArn(topicName)).
                    withEndpoint(getQueueArn(queueName)));

    amazonSNS.publish(new PublishRequest(makeTopicArn(topicName), message));

    ReceiveMessageResult result = _amazonSQS2.receiveMessage(new ReceiveMessageRequest(queueUrl).
            withWaitTimeSeconds(15));
    Assert.assertEquals(result.getMessages().size(), 1);
    Assert.assertEquals(result.getMessages().get(0).getBody(), message);
}

@Test
public void testGetRequest() throws Exception {
    String targetArn = "arn:aws:sns:us-east-1:123456789012:endpoint/APNS/64d65215-f7d4-4ebc-8c45-c1237510e9dc/df7feec2-1aca-3f4f-a286-64e0c1427161";
    String topicArn = "arn:aws:sns:us-east-1:970456703086:Blah";
    String message = "message";
    String subject = "subject";
    String messageStruct = "messageStructure";
    PublishRequest r = new PublishRequestBuilder(message)
            .message(message)
            .messageStructure(messageStruct)
            .subject(subject)
            .targetArn(targetArn)
            .topicArn(topicArn)
            .getRequest();
    assertEquals(message, r.getMessage());
    assertEquals(messageStruct, r.getMessageStructure());
    assertEquals(subject, r.getSubject());
    assertEquals(targetArn, r.getTargetArn());
    assertEquals(topicArn, r.getTopicArn());
}

/**
 * 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 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());
}

@Test
public void failureBeforeTableReplicationStartIsCalled() {
  SnsListener listener = new SnsListener(client, config, clock);
  listener.circusTrainStartUp(new String[] {}, sourceCatalog, replicaCatalog);
  listener.tableReplicationFailure(tableReplication, EVENT_ID, ERROR);

  verify(client, times(1)).publish(requestCaptor.capture());
  PublishRequest request = requestCaptor.getValue();
  assertThat(request.getSubject(), is(SUBJECT));
  assertThat(request.getTopicArn(), is("failArn"));
  assertThat(request.getMessage(), is("{\"protocolVersion\":\"1.0\",\"type\":\"FAILURE\",\"headers\":"
      + "{\"pipeline-id\":\"0943879438\"},\"startTime\":\"starttime\",\"endTime\":\"endtime\",\"eventId\":"
      + "\"EVENT_ID\",\"sourceCatalog\":\"sourceCatalogName\",\"replicaCatalog\":\"replicaCatalogName\","
      + "\"sourceTable\":\"srcDb.srcTable\",\"replicaTable\":\"replicaDb.replicaTable\",\"bytesReplicated\":0,\"errorMessage\":\"error message\"}"));
}

/**
 * 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);
    }
}

@Override
public void sendSmsAsync(SmsRequest smsRequest) {
    try {
        this.asyncSNS.publishAsync(new PublishRequest()
                .withMessage(smsRequest.getMessage())
                .withPhoneNumber(smsRequest.getCountryCode() + smsRequest.getPhoneNumber())
                .withMessageAttributes(this.smsAttributes), this.asyncHandler);
    } catch (Exception ex) {
        LOGGER.error("Exception while sending SMS asynchronously!!", 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";
}

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);
}