Java 类com.amazonaws.services.sns.AmazonSNS 实例源码

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

@Test
public void testGetAmazonSNSClientCacheHitMiss()
{
    // Create an AWS parameters DTO that contains proxy information.
    AwsParamsDto awsParamsDto = new AwsParamsDto(NO_AWS_ACCESS_KEY, NO_AWS_SECRET_KEY, NO_SESSION_TOKEN, HTTP_PROXY_HOST, HTTP_PROXY_PORT);

    // Get an Amazon SNS client.
    AmazonSNS amazonSNS = awsClientFactory.getAmazonSNSClient(awsParamsDto);

    // Confirm a cache hit.
    assertEquals(amazonSNS,
        awsClientFactory.getAmazonSNSClient(new AwsParamsDto(NO_AWS_ACCESS_KEY, NO_AWS_SECRET_KEY, NO_SESSION_TOKEN, HTTP_PROXY_HOST, HTTP_PROXY_PORT)));

    // Confirm a cache miss due to http proxy information.
    assertNotEquals(amazonSNS, awsClientFactory
        .getAmazonSNSClient(new AwsParamsDto(NO_AWS_ACCESS_KEY, NO_AWS_SECRET_KEY, NO_SESSION_TOKEN, HTTP_PROXY_HOST_2, HTTP_PROXY_PORT_2)));

    // Clear the cache.
    cacheManager.getCache(DaoSpringModuleConfig.HERD_CACHE_NAME).clear();

    // Confirm a cache miss due to cleared cache.
    assertNotEquals(amazonSNS, awsClientFactory.getAmazonSNSClient(awsParamsDto));
}

private void retryVerifyPlatformApplication(AmazonSNS client) {
    int retry_count = 10000;
    while(true) {
        try {
            verifyPlatformApplication(this.client);
            break;
        } catch (Exception e) {
            retry_count--;
            System.out.println("Exception caught. message:" + e.getMessage());
            System.out.println(">>>>> retry count remaining: "+retry_count);
            if (retry_count == 0) {
                System.exit(BatchCreatePlatformEndpointSample.NOT_FOUND_ERROR_CODE);
            } else {
                try {
                    Thread.sleep(3000); 
                } catch (InterruptedException ee) {
                    System.exit(BatchCreatePlatformEndpointSample.NOT_FOUND_ERROR_CODE);
                }
                continue;
            }
        }
    }
}

@Override
public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
    List<BeanDefinition> methodArgumentResolvers = new ManagedList<>(3);

    BeanDefinitionBuilder notificationMessageHandler = BeanDefinitionBuilder.rootBeanDefinition(NotificationMessageHandlerMethodArgumentResolver.class);
    methodArgumentResolvers.add(notificationMessageHandler.getBeanDefinition());

    BeanDefinitionBuilder notificationSubjectHandler = BeanDefinitionBuilder.rootBeanDefinition(NotificationSubjectHandlerMethodArgumentResolver.class);
    methodArgumentResolvers.add(notificationSubjectHandler.getBeanDefinition());

    BeanDefinitionBuilder notificationStatusBeanDefinition = BeanDefinitionBuilder.rootBeanDefinition(NotificationStatusHandlerMethodArgumentResolver.class);
    notificationStatusBeanDefinition.addConstructorArgReference(AmazonWebserviceClientConfigurationUtils.getBeanName(AmazonSNS.class.getName()));
    methodArgumentResolvers.add(notificationStatusBeanDefinition.getBeanDefinition());

    BeanDefinition requestMappingHandlerAdapter = beanFactory.getBeanDefinition("requestMappingHandlerAdapter");

    requestMappingHandlerAdapter.getPropertyValues().addPropertyValue("customArgumentResolvers", methodArgumentResolvers);
}

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 resolveDestination_withExistentTopic_returnsTopicArnFoundWhileListingTopic() throws Exception {
    // Arrange
    String topicArn = "arn:aws:sns:eu-west:123456789012:test";

    AmazonSNS sns = mock(AmazonSNS.class);
    when(sns.listTopics(new ListTopicsRequest(null))).thenReturn(new ListTopicsResult().withTopics(new Topic().withTopicArn(topicArn)));

    DynamicTopicDestinationResolver resolver = new DynamicTopicDestinationResolver(sns);

    // Act
    String resolvedDestinationName = resolver.resolveDestination("test");

    // Assert
    assertEquals(topicArn, resolvedDestinationName);
}

@Test
public void resolveDestination_withExistentTopicAndMarker_returnsTopicArnFoundWhileListingTopic() throws Exception {
    // Arrange

    AmazonSNS sns = mock(AmazonSNS.class);
    when(sns.listTopics(new ListTopicsRequest(null))).thenReturn(new ListTopicsResult().withNextToken("mark"));

    String topicArn = "arn:aws:sns:eu-west:123456789012:test";
    when(sns.listTopics(new ListTopicsRequest("mark"))).thenReturn(new ListTopicsResult().withTopics(new Topic().withTopicArn(topicArn)));

    DynamicTopicDestinationResolver resolver = new DynamicTopicDestinationResolver(sns);

    // Act
    String resolvedDestinationName = resolver.resolveDestination("test");

    // Assert
    assertEquals(topicArn, resolvedDestinationName);
}

@Test
public void resolveDestination_withAutoCreateEnabled_shouldCreateTopicDirectly() throws Exception {
    // Arrange
    String topicArn = "arn:aws:sns:eu-west:123456789012:test";

    AmazonSNS sns = mock(AmazonSNS.class);
    when(sns.createTopic(new CreateTopicRequest("test"))).thenReturn(new CreateTopicResult().withTopicArn(topicArn));

    DynamicTopicDestinationResolver resolver = new DynamicTopicDestinationResolver(sns);
    resolver.setAutoCreate(true);

    // Act
    String resolvedDestinationName = resolver.resolveDestination("test");

    // Assert
    assertEquals(topicArn, resolvedDestinationName);
}

@Test
public void resolveDestination_withResourceIdResolver_shouldCallIt() throws Exception {
    // Arrange
    String physicalTopicName = "arn:aws:sns:eu-west:123456789012:myTopic";
    String logicalTopicName = "myTopic";

    ResourceIdResolver resourceIdResolver = mock(ResourceIdResolver.class);
    when(resourceIdResolver.resolveToPhysicalResourceId(logicalTopicName)).thenReturn(physicalTopicName);

    AmazonSNS sns = mock(AmazonSNS.class);
    when(sns.listTopics(new ListTopicsRequest(null))).thenReturn(new ListTopicsResult().withTopics(new Topic().withTopicArn(physicalTopicName)));

    DynamicTopicDestinationResolver resolver = new DynamicTopicDestinationResolver(sns, resourceIdResolver);

    // Assert
    String resolvedDestinationName = resolver.resolveDestination(logicalTopicName);

    // Assert
    assertEquals(physicalTopicName, resolvedDestinationName);
}

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

@Test
public void resolveArgument_wrongMessageType_reportsErrors() throws Exception {
    //Arrange
    this.expectedException.expect(IllegalArgumentException.class);
    this.expectedException.expectMessage("NotificationStatus is only available");

    AmazonSNS amazonSns = mock(AmazonSNS.class);
    NotificationStatusHandlerMethodArgumentResolver resolver = new NotificationStatusHandlerMethodArgumentResolver(amazonSns);

    byte[] subscriptionRequestJsonContent = FileCopyUtils.copyToByteArray(new ClassPathResource("notificationMessage.json", getClass()).getInputStream());
    MockHttpServletRequest servletRequest = new MockHttpServletRequest();
    servletRequest.setContent(subscriptionRequestJsonContent);

    MethodParameter methodParameter = new MethodParameter(ReflectionUtils.findMethod(NotificationMethods.class, "subscriptionMethod", NotificationStatus.class), 0);

    //Act
    resolver.resolveArgument(methodParameter, null, new ServletWebRequest(servletRequest), null);

    //Assert
}

@Test
public void resolveArgument_subscriptionRequest_createsValidSubscriptionStatus() throws Exception {
    //Arrange
    AmazonSNS amazonSns = mock(AmazonSNS.class);
    NotificationStatusHandlerMethodArgumentResolver resolver = new NotificationStatusHandlerMethodArgumentResolver(amazonSns);

    byte[] subscriptionRequestJsonContent = FileCopyUtils.copyToByteArray(new ClassPathResource("subscriptionConfirmation.json", getClass()).getInputStream());

    MockHttpServletRequest servletRequest = new MockHttpServletRequest();
    servletRequest.setContent(subscriptionRequestJsonContent);

    MethodParameter methodParameter = new MethodParameter(ReflectionUtils.findMethod(NotificationMethods.class, "subscriptionMethod", NotificationStatus.class), 0);

    //Act
    Object resolvedArgument = resolver.resolveArgument(methodParameter, null, new ServletWebRequest(servletRequest), null);

    //Assert
    assertTrue(resolvedArgument instanceof NotificationStatus);
    ((NotificationStatus) resolvedArgument).confirmSubscription();
    verify(amazonSns, times(1)).confirmSubscription("arn:aws:sns:eu-west-1:111111111111:mySampleTopic", "111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111");
}

private static AmazonSNS getSNS() throws IOException {
    String awsAccessKey = System.getProperty("AWS_ACCESS_KEY_ID"); // "YOUR_AWS_ACCESS_KEY";
    String awsSecretKey = System.getProperty("AWS_SECRET_KEY"); // "YOUR_AWS_SECRET_KEY";

    if (awsAccessKey == null)
        awsAccessKey = AWS_ACCESS_KEY;
    if (awsSecretKey == null)
        awsSecretKey = AWS_SECRET_KEY;

    AWSCredentials credentials = new BasicAWSCredentials(awsAccessKey,
            awsSecretKey);
    AmazonSNS sns = new AmazonSNSClient(credentials);

    sns.setEndpoint("https://sns.us-west-2.amazonaws.com");
    return sns;
}

/**
 * Sets up a queue so that it subscribes to all messages that gets published on the
 * specified topic. The queue and topic must be created before-hand, or an exception
 * will be thrown.
 *
 * This method will also configure the queue so that it has permissions to receive
 * messages from the topic.
 *
 * @param queueURL The queue that should receive the messages posted to the topic.
 * @param topicARN The topic whose messages should be posted to the queue.
 * @param amazonSQS Amazon SQS client.
 * @param amazonSNS Amazon SNS client.
 */
public static void subscribeSQSQueueToSNSTopic(
    String queueURL,
    String topicARN,
    AmazonSQS amazonSQS,
    AmazonSNS amazonSNS
) {
    // Verify that the queue exists, and get its ARN
    String queueARN = getSQSQueueARN(amazonSQS, queueURL);

    // The "is already subscribing?"-check has a dual purpose: it will also verify that the
    // topic is already created. If the topic is not created beforehand, this will throw a
    // com.amazonaws.services.sns.model.NotFoundException
    if (isSQSQueueSubscribingToSNSTopic(amazonSNS, queueARN, topicARN)) {
        // Subscription already configured. Do nothing.
        return;
    }

    allowSQSQueueToReceiveMessagesFromSNSTopic(amazonSQS, queueURL, queueARN, topicARN);
    subscribeSQSQueueToSNSTopic(amazonSNS, queueARN, topicARN);
}

private static boolean isSQSQueueSubscribingToSNSTopic(
    AmazonSNS amazonSNS,
    String queueARN,
    String topicARN
) {
    // This statement will throw if the topic does not exist.
    ListSubscriptionsByTopicResult subscriptions = amazonSNS.listSubscriptionsByTopic(
        new ListSubscriptionsByTopicRequest()
            .withTopicArn(topicARN)
    );
    for (Subscription subscription : subscriptions.getSubscriptions()) {
        if (subscription.getEndpoint().equals(queueARN)) {
            return true;
        }
    }
    return false;
}

public static MessagePublisher encryptedPublisherService(
        AmazonSNS client,
        final String topicArn,
        final byte[] publicPgpKey
) throws MessagingException {

    if (empty(publicPgpKey)) {
        throw new MessagingException("Can't create encryptedPublisherService with null as public PGP key");
    }

    MessageSerializer messageSerializer;
    try {
        messageSerializer = new DefaultMessageSerializer(publicPgpKey);
    } catch (CryptographyException e) {
        throw new MessagingException("Failed to load public PGP key needed to encrypt messages.", e);
    }
    return new PublisherService(client, topicArn, messageSerializer);
}

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

@Override
protected void createAWSClient()
{
    client = tryClientFactory(config.clientFactoryMethod, AmazonSNS.class, true);
    if ((client == null) && (config.clientEndpoint == null))
    {
        client = tryClientFactory("com.amazonaws.services.sns.AmazonSNSClientBuilder.defaultClient", AmazonSNS.class, false);
    }
    if (client == null)
    {
        LogLog.debug(getClass().getSimpleName() + ": creating service client via constructor");
        client = tryConfigureEndpointOrRegion(new AmazonSNSClient(), config.clientEndpoint);
    }
}

/**
 *  Creates a client proxy outside of the factory.
 */
public AmazonSNS createClient()
{
    return (AmazonSNS)Proxy.newProxyInstance(
                                getClass().getClassLoader(),
                                new Class<?>[] { AmazonSNS.class },
                                MockSNSClient.this);
}

@Test
public void snsClient() {
  AWSCredentialsProvider credentialsProvider = mock(AWSCredentialsProvider.class);
  when(credentialsProvider.getCredentials()).thenReturn(new BasicAWSCredentials("accessKey", "secretKey"));
  ListenerConfig config = new ListenerConfig();
  config.setRegion("eu-west-1");
  AmazonSNS sns = configuration.amazonSNS(config, credentialsProvider);
  assertThat(sns, is(not(nullValue())));
}

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

/**
 * Returns an SNS client.
 *
 * <p>This method is provided so unit tests can mock out SNS.
 */
protected AmazonSNS getSNSClient() {

  // Use a getter here so unit tests can substitute a mock client
  AmazonSNS client = AmazonSNSClientBuilder.standard().withRegion(region.getName()).build();
  return client;
}

/**
 * Returns an SNS client.
 *
 * <p>This method is provided so unit tests can mock out SNS.
 */
protected AmazonSNS getSNSClient() {

  // Use a getter here so unit tests can substitute a mock client
  AmazonSNS client = AmazonSNSClientBuilder.standard().withRegion(region.getName()).build();
  return client;
}

/**
 * Returns an SNS client.
 *
 * <p>This method is provided so unit tests can mock out SNS.
 */
protected AmazonSNS getSNSClient() {

  // Use a getter here so unit tests can substitute a mock client
  AmazonSNS client = AmazonSNSClientBuilder.standard().withRegion(region.getName()).build();
  return client;
}

/**
 * Returns an SNS client.
 *
 * <p>This method is provided so unit tests can mock out SNS.
 */
protected AmazonSNS getSNSClient() {

  // Use a getter here so unit tests can substitute a mock client
  AmazonSNS client = AmazonSNSClientBuilder.standard().withRegion(region.getName()).build();
  return client;
}

@Test
public void testRefreshAllPagesThrowsWhenS3Throws() throws Exception {
  // ARRANGE
  thrown.expect(Exception.class);
  thrown.expectMessage("Exception caught while copying booking page to S3");

  initialisePageManager();

  // Make S3 throw:
  // Transfer interface is implemented by Uploads, Downloads, and Copies
  Transfer mockTransfer = mockery.mock(Transfer.class);
  mockery.checking(new Expectations() {
    {
      allowing(mockTransfer).isDone();
      will(returnValue(true));
      allowing(mockTransfer).waitForCompletion();
    }
  });
  mockTransferManager = mockery.mock(IS3TransferManager.class);
  mockery.checking(new Expectations() {
    {
      oneOf(mockTransferManager).upload(with(any(PutObjectRequest.class)));
      will(throwException(new AmazonServiceException("Grrr...")));
      // Should throw before copy is called
      never(mockTransferManager).copy(with(any(CopyObjectRequest.class)));
    }
  });
  pageManager.setS3TransferManager(mockTransferManager);

  mockSNSClient = mockery.mock(AmazonSNS.class);
  mockery.checking(new Expectations() {
    {
      ignoring(mockSNSClient);
    }
  });
  pageManager.setSNSClient(mockSNSClient);

  // ACT - this should throw
  pageManager.refreshAllPages(validDates, apiGatewayBaseUrl, revvingSuffix);
}

@Test
public void testApplyRulesThrowsWhenTheOptimisticPersisterThrows() throws Exception {

  // ARRANGE
  thrown.expect(Exception.class);
  String message = "Test OptimisticPersister exception";
  thrown.expectMessage(message);

  initialiseRuleManager();

  mockery.checking(new Expectations() {
    {
      oneOf(mockOptimisticPersister).get(with(equal(ruleItemName)));
      will(throwException(new Exception(message)));
    }
  });

  // Set up mock SNS client
  mockSNSClient = mockery.mock(AmazonSNS.class);
  mockery.checking(new Expectations() {
    {
      ignoring(mockSNSClient);
    }
  });
  ruleManager.setSNSClient(mockSNSClient);

  // ACT
  // This should throw
  ruleManager
      .applyRules(existingSaturdayRecurringRuleWithExclusion.getBooking().getDate(), false);
}

@Test
public void testApplyRulesThrowsWhenTheBookingManagerThrows() throws Exception {

  // ARRANGE
  thrown.expect(Exception.class);
  String message = "Test BookingManager exception";
  thrown.expectMessage(message);

  initialiseRuleManager();
  expectOptimisticPersisterToReturnVersionedAttributes(42);

  mockery.checking(new Expectations() {
    {
      oneOf(mockBookingManager).createBooking(with(anything()), with.booleanIs(anything()));
      will(throwException(new Exception(message)));
    }
  });

  // Set up mock SNS client
  mockSNSClient = mockery.mock(AmazonSNS.class);
  mockery.checking(new Expectations() {
    {
      ignoring(mockSNSClient);
    }
  });
  ruleManager.setSNSClient(mockSNSClient);

  // ACT
  // This should throw
  ruleManager
      .applyRules(existingSaturdayRecurringRuleWithExclusion.getBooking().getDate(), false);
}

@Test
public void testApplyRulesNotifiesTheSnsTopicWhenItThrows() throws Exception {
  // It is useful for the admin user to be notified whenever the application
  // of booking rules does not succeed - so that they can apply rule bookings
  // manually instead. This tests that whenever the rule manager catches an
  // exception while applying rules, it notifies the admin SNS topic.

  // ARRANGE
  thrown.expect(Exception.class);
  String message = "Test BookingManager exception";
  thrown.expectMessage(message);

  initialiseRuleManager();
  expectOptimisticPersisterToReturnVersionedAttributes(42);

  mockery.checking(new Expectations() {
    {
      oneOf(mockBookingManager).createBooking(with(anything()), with.booleanIs(anything()));
      will(throwException(new Exception(message)));
    }
  });

  // Set up mock SNS client to expect a notification
  mockSNSClient = mockery.mock(AmazonSNS.class);
  String partialMessage = "Apologies - but there was an error applying the booking rules";
  mockery.checking(new Expectations() {
    {
      oneOf(mockSNSClient).publish(with(equal(adminSnsTopicArn)),
          with(startsWith(partialMessage)), with(equal("Sqawsh booking rules failed to apply")));
    }
  });
  ruleManager.setSNSClient(mockSNSClient);

  // ACT
  // This should throw - and notify the SNS topic
  ruleManager
      .applyRules(existingSaturdayRecurringRuleWithExclusion.getBooking().getDate(), false);
}

@Override
public AmazonSNS get(SnsModule.EndpointConfig config) {
    AmazonSNSClientBuilder builder = AmazonSNSClientBuilder.standard();
    builder.withCredentials(credentialsProvider);
    if (config.endpoint().isPresent()) {
        builder.withEndpointConfiguration(
            new AwsClientBuilder.EndpointConfiguration(config.getEndpoint(), config.getRegionName())
        );
    } else {
        builder.withRegion(config.getRegionName());
    }
    return builder.build();
}

@Provides
@Singleton
protected AmazonSNS provideAmazonSNS(Region region, AWSCredentialsProvider credentialsProvider) {
    AmazonSNS amazonSNS = new AmazonSNSClient(credentialsProvider);
    amazonSNS.setRegion(region);
    return amazonSNS;
}

@Provides
@Singleton
protected Optional<SNSStashStateListener> provideSNSStashStateListener(AmazonSNS amazonSNS, Environment environment,
                                                                       PluginServerMetadata metadata) {
    String snsTopic = _config.getNotifications().getSnsTopic();
    if (snsTopic != null) {
        SNSStashStateListener listener = new SNSStashStateListener(amazonSNS, snsTopic);
        listener.init(environment, metadata, null);
        return Optional.of(listener);
    }
    return Optional.absent();
}

private void init() {
    AWSCredentials awsCredentials = new BasicAWSCredentials(configuration.getAccessKeyId(), configuration.getSecretAccessKey());
    AWSStaticCredentialsProvider credProvider = new AWSStaticCredentialsProvider(awsCredentials);
    AmazonSNS sns = AmazonSNSClient.builder()
            .withCredentials(credProvider)
            .withRegion(configuration.getRegion())
            .build();
    this.snsMessageHandler = new SnsMessageHandler(sns);

}