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

@Test
public void testCreateListDeleteTopic_shouldCreateReturnAndDelete() {
  mockSns(new MockParameters());

  ListTopicsResult listTopicResultBefore = sns.listTopics();
  assertEquals("topic list should contain zero items before insert",0,listTopicResultBefore.getTopics().size());

  CreateTopicRequest create = new CreateTopicRequest()
      .withName("major-topic");
  CreateTopicResult createResult = sns.createTopic(create);
  String topicArn = createResult.getTopicArn();
  assertNotNull("verify returned topic ARN", topicArn);

  ListTopicsResult listTopicResult = sns.listTopics();
  assertEquals("after insert topic list should contain 1 item",1,listTopicResult.getTopics().size());
  assertEquals("after insert topic list should contain before inserted topic arn", topicArn, 
      listTopicResult.getTopics().get(0).getTopicArn());

  DeleteTopicResult deleteResult = sns.deleteTopic(topicArn);
  assertNotNull(deleteResult);

  ListTopicsResult listTopicsAfterDeletion = sns.listTopics();
  assertEquals("topic list should contain zero items after deletion",0,listTopicsAfterDeletion.getTopics().size());
}

@Test
public void testSetGetSubscriptionAttributes_shouldSetAndRespondSubscriptionAttributes() {
  mockSns(new MockParameters());
  // create topic and subscription
  CreateTopicResult topicResult = sns.createTopic(new CreateTopicRequest().withName("important-topic"));
  SubscribeResult subscribeResult = sns.subscribe(new SubscribeRequest().withTopicArn(topicResult.getTopicArn())
      .withProtocol("sqs").withEndpoint("arn:aws:sqs:us-east-1:123456789012:queue1"));
  // set subscription attribute
  SetSubscriptionAttributesResult setAttrResult =  sns.setSubscriptionAttributes(new SetSubscriptionAttributesRequest()
      .withAttributeName("unicorns-exist").withAttributeValue("only in scotland").withSubscriptionArn(subscribeResult.getSubscriptionArn()));
  assertNotNull("verify setting attributes result", setAttrResult);
  // get subscription attribute
  GetSubscriptionAttributesResult subAttributes = sns.getSubscriptionAttributes(new GetSubscriptionAttributesRequest()
      .withSubscriptionArn(subscribeResult.getSubscriptionArn()));
  assertEquals("verify subscription attribute","only in scotland",subAttributes.getAttributes().get("unicorns-exist"));
}

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

/**
 * 
 * @param email
 * @return
 */
public String getOrCreateNotification(String email) {
    String ret = null;
    String topicName = getTopicName(email);
    String nextToken = null;
    do {
        ListTopicsResult listTopics = asyncSnsClient.listTopics(nextToken);
        List<Topic> topics = listTopics.getTopics();
        for (Topic s : topics) {
            if (s.getTopicArn().endsWith(topicName)) {
                ret = s.getTopicArn();
                break;
            }
        }
        nextToken = listTopics.getNextToken();
    } while (ret == null && nextToken != null);
    if (ret == null) {
        // create the topic and the subscription
        CreateTopicResult topic = asyncSnsClient.createTopic(topicName);
        SubscribeRequest req = new SubscribeRequest(topic.getTopicArn(), "email", email);
        asyncSnsClient.subscribeAsync(req);
        ret = topic.getTopicArn();
    }

    return ret;
}

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

private void createTopicIfNotExists() {
    for (Topic topic : client.listTopics().getTopics()) {
        if (topic.getTopicArn().contains(topicName)) {
            topicArn = topic.getTopicArn();
            break;
        }
    }
    if (topicArn == null) {
        CreateTopicRequest request = new CreateTopicRequest(topicName);
        CreateTopicResult result = client.createTopic(request);
        topicArn = result.getTopicArn();
        log.debug("Topic created, arn: " + topicArn);
    } else {
        log.debug("Topic already created: " + topicArn);
    }
}

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

    MockSNSClient mockClient = new MockSNSClient("example", Arrays.asList("argle", "bargle"))
    {
        @Override
        protected CreateTopicResult createTopic(String name)
        {
            throw new TestingException("arbitrary failure");
        }
    };

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

    // first message triggers writer creation

    logger.info("message one");
    waitForInitialization();

    AbstractLogWriter writer = (AbstractLogWriter)appender.getWriter();
    MessageQueue messageQueue = appender.getMessageQueue();

    assertTrue("initialization message was non-blank",  ! writer.getInitializationMessage().equals(""));
    assertEquals("initialization exception retained",   TestingException.class,     writer.getInitializationException().getClass());
    assertEquals("message queue set to discard all",    0,                          messageQueue.getDiscardThreshold());
    assertEquals("message queue set to discard all",    DiscardAction.oldest,       messageQueue.getDiscardAction());
    assertEquals("messages in queue (initial)",         1,                          messageQueue.toList().size());

    // trying to log another message should clear the queue

    logger.info("message two");
    assertEquals("messages in queue (second try)",      0,                          messageQueue.toList().size());
}

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

@Test
public void testGetSetTopicAttributes_shouldAddAndRespondAttributes() {
  mockSns(new MockParameters());
  CreateTopicResult topicResult = sns.createTopic(new CreateTopicRequest().withName("attributefull-topic"));

  SetTopicAttributesResult setAttrResult = sns.setTopicAttributes(topicResult.getTopicArn(), "planet", "Omega 3");
  assertNotNull(setAttrResult);

  GetTopicAttributesResult topicAttributes = sns.getTopicAttributes(new GetTopicAttributesRequest().withTopicArn(topicResult.getTopicArn()));

  assertEquals("verify added attribute is correct", "Omega 3", topicAttributes.getAttributes().get("planet"));

  sns.deleteTopic(topicResult.getTopicArn());
}

private Supplier<String> createTopicSupplier(final String topicName) {
    return Suppliers.memoize(new Supplier<String>() {
        @Override
        public String get() {
            // This will create the topic if it doesn't exist or return the existing topic if it does.
            CreateTopicResult topic = _amazonSNS.createTopic(topicName);
            return topic.getTopicArn();
        }
    });
}

@Override
public void subscribeForFacility(Long equipmentnumber, String email) {
    Validate.notNull(equipmentnumber);
    final Facility facility = getFacilityService().findByEquipmentnumber(equipmentnumber);
    if (facility != null) {
        String topicName = MessageFormat.format(FACILITY_TOPIC_NAME_PATTERN, equipmentnumber);
        CreateTopicResult topicResult = getClient().createTopic(topicName);
        getClient().subscribe(new SubscribeRequest(topicResult.getTopicArn(), EMAIL_PROTOCOL, email));
    }
}

@Override
public void doStart() throws Exception {
    super.doStart();
    snsClient = configuration.getAmazonSNSClient() != null
        ? configuration.getAmazonSNSClient() : createSNSClient();

    // Override the setting Endpoint from url
    if (ObjectHelper.isNotEmpty(configuration.getAmazonSNSEndpoint())) {
        LOG.trace("Updating the SNS region with : {} " + configuration.getAmazonSNSEndpoint());
        snsClient.setEndpoint(configuration.getAmazonSNSEndpoint());
    }

    if (configuration.getTopicArn() == null) {
        // creates a new topic, or returns the URL of an existing one
        CreateTopicRequest request = new CreateTopicRequest(configuration.getTopicName());

        LOG.trace("Creating topic [{}] with request [{}]...", configuration.getTopicName(), request);

        CreateTopicResult result = snsClient.createTopic(request);
        configuration.setTopicArn(result.getTopicArn());

        LOG.trace("Topic created with Amazon resource name: {}", configuration.getTopicArn());
    }

    if (ObjectHelper.isNotEmpty(configuration.getPolicy())) {
        LOG.trace("Updating topic [{}] with policy [{}]", configuration.getTopicArn(), configuration.getPolicy());

        snsClient.setTopicAttributes(new SetTopicAttributesRequest(configuration.getTopicArn(), "Policy", configuration.getPolicy()));

        LOG.trace("Topic policy updated");
    }

}

@Test
public void createTopic() {
    CreateTopicResult result = mock(CreateTopicResult.class);
    when(sns.createTopic(anyString())).thenReturn(result);

    assertThat(service.createTopic(TOPIC_NAME)).isEqualTo(result);
    verify(sns).createTopic(TOPIC_NAME);
}

@Override
public void subscribeForFacility(Long equipmentnumber, String email) {
    Validate.notNull(equipmentnumber);
    final Facility facility = getFacilityService().findByEquipmentnumber(equipmentnumber);
    if (facility != null) {
        String topicName = MessageFormat.format(FACILITY_TOPIC_NAME_PATTERN, equipmentnumber);
        CreateTopicResult topicResult = getClient().createTopic(topicName);
        getClient().subscribe(new SubscribeRequest(topicResult.getTopicArn(), EMAIL_PROTOCOL, email));
    }
}

@Override
public Topic createTopic(CreateTopicRequest request,
        ResultCapture<CreateTopicResult> extractor) {

    ActionResult result = service.performAction("CreateTopic", request,
            extractor);

    if (result == null) return null;
    return new TopicImpl(result.getResource());
}

@Override
public Topic createTopic(String name, ResultCapture<CreateTopicResult>
        extractor) {

    CreateTopicRequest request = new CreateTopicRequest()
        .withName(name);
    return createTopic(request, extractor);
}

@Override
public CreateTopicResult createTopic(CreateTopicRequest createTopicRequest) throws AmazonClientException {
    String topicArn = BASE_ARN + createTopicRequest.getName();
    CreateTopicResult result = new CreateTopicResult().withTopicArn(topicArn);
    if (!_subscriptionsForTopic.containsKey(topicArn)) {
        _subscriptionsForTopic.put(topicArn, new ArrayList<String>());
    }
    return result;
}

/**
 * Creates a SNS topic.
 *
 * @param topic string
 * @param email address to be notified when there is any event for this topic.
 * @return false if there is any error.
 * @throws Exception
 */
public boolean createTopic(final String topic, final String email)
        throws Exception{
    CreateTopicRequest request = new CreateTopicRequest()
        .withName(topic);
    CreateTopicResult result = null;

    try {
        result = snsClient.createTopic(request);
    } catch (Exception e) {
        LGR.setUseParentHandlers(true);
        LGR.log(Level.SEVERE, null, e);
        LGR.setUseParentHandlers(false);
        return false;
    }

    topicARN = result.getTopicArn();
    LGR.log(Level.INFO, "topic arn is {0}", topicARN);

    SubscribeRequest request2 = new SubscribeRequest()
        .withTopicArn(topicARN).withEndpoint(email).withProtocol("email");
    SubscribeResult result2 = snsClient.subscribe(request2);

    LGR.log(Level.INFO, "subscription ARN {0}",
            result2.getSubscriptionArn());     

    return true;
}

@Test
public void shouldFindSNSTopicIfPresent() {
    CreateTopicResult createResult = snsClient.createTopic(SNSNotificationSender.TOPIC_NAME);
    String arn = createResult.getTopicArn();

    assertFalse(sender.getTopicARN().isEmpty());

    snsClient.deleteTopic(arn);

    SNSNotificationSender senderB = new SNSNotificationSender(snsClient);
    assertTrue(senderB.getTopicARN().isEmpty());
}

@Test
public void testSubscribeConfirmListUnsubscribe_shouldCreateVerifyListAndRemoveSubscription() {
  mockSns(new MockParameters());
  // create topic
  CreateTopicResult topicResult = sns.createTopic(new CreateTopicRequest().withName("important-topic"));

  // subscribe to first topic
  SubscribeResult subscribeResult = sns.subscribe(new SubscribeRequest().withTopicArn(topicResult.getTopicArn())
      .withProtocol("sqs").withEndpoint("arn:aws:sqs:us-east-1:123456789012:queue1"));
  assertNotNull("verify subscription ARN is created", subscribeResult.getSubscriptionArn());

  // create second topic and subscribe to that one
  CreateTopicResult secondTopicResult = sns.createTopic(new CreateTopicRequest().withName("someOther-topic"));
  sns.subscribe(new SubscribeRequest().withTopicArn(secondTopicResult.getTopicArn())
      .withProtocol("sqs").withEndpoint("arn:aws:sqs:us-east-1:564654654:queue7"));

  // confirm first subscription
  ConfirmSubscriptionResult confirmResultAuth = sns.confirmSubscription(new ConfirmSubscriptionRequest()
      .withAuthenticateOnUnsubscribe("true").withToken("gold-token").withTopicArn(topicResult.getTopicArn()));
  assertNotNull("verify auth confirmation with responded topic arn", confirmResultAuth.getSubscriptionArn());
  ConfirmSubscriptionResult confirmResultNoAuth = sns.confirmSubscription(new ConfirmSubscriptionRequest()
      .withAuthenticateOnUnsubscribe("false").withToken("gold-token").withTopicArn(topicResult.getTopicArn()));
  assertNotNull("verify no auth confirmation with responded topic arn", confirmResultNoAuth.getSubscriptionArn());

  // list all subscriptions
  ListSubscriptionsResult allSubs = sns.listSubscriptions();
  assertEquals("verify correct total subscription count", 2, allSubs.getSubscriptions().size());
  com.amazonaws.services.sns.model.Subscription firstSub = allSubs.getSubscriptions().stream().filter(sub -> sub.getTopicArn().equals(topicResult.getTopicArn())).findFirst().get();
  assertEquals("verify the correct subscription arn", subscribeResult.getSubscriptionArn(), firstSub.getSubscriptionArn());
  assertEquals("verify the correct subscription topic", topicResult.getTopicArn(), firstSub.getTopicArn());
  assertEquals("verify the correct subscription protocol", "sqs", firstSub.getProtocol());
  assertEquals("verify the correct subscription endpoint", "arn:aws:sqs:us-east-1:123456789012:queue1", firstSub.getEndpoint());
  assertNotNull("verify the correct subscription owner", firstSub.getOwner());

  // list subscriptions of first topic
  ListSubscriptionsByTopicResult topicsSubscriptions = sns.listSubscriptionsByTopic(new ListSubscriptionsByTopicRequest(topicResult.getTopicArn()));
  assertEquals("verify that the one subscription is contained in list", 1, topicsSubscriptions.getSubscriptions().size());
  assertEquals("verify the correct subscription arn", subscribeResult.getSubscriptionArn(), topicsSubscriptions.getSubscriptions().get(0).getSubscriptionArn());
  assertEquals("verify the correct subscription topic", topicResult.getTopicArn(), topicsSubscriptions.getSubscriptions().get(0).getTopicArn());
  assertEquals("verify the correct subscription protocol", "sqs", topicsSubscriptions.getSubscriptions().get(0).getProtocol());
  assertEquals("verify the correct subscription endpoint", "arn:aws:sqs:us-east-1:123456789012:queue1", topicsSubscriptions.getSubscriptions().get(0).getEndpoint());
  assertNotNull("verify the correct subscription owner", topicsSubscriptions.getSubscriptions().get(0).getOwner());

  // unsubscribe first topic
  assertNotNull(sns.unsubscribe(subscribeResult.getSubscriptionArn()));

  // check if really unsubscribed
  ListSubscriptionsByTopicResult subsToFirstTopicAfterUnsubscribe = sns.listSubscriptionsByTopic(new ListSubscriptionsByTopicRequest(topicResult.getTopicArn()));
  assertEquals("topic should be gone", 0, subsToFirstTopicAfterUnsubscribe.getSubscriptions().size());

  // cleanup
  sns.deleteTopic(topicResult.getTopicArn());
  sns.deleteTopic(secondTopicResult.getTopicArn());
}

private String createTopic(String arn) {
    CreateTopicResult result
            = snsClient.createTopic(new CreateTopicRequest(arn));
    return result.getTopicArn();
}

@Override
public void subscribeForAllFacilities(String email) {
    CreateTopicResult topicResult = getClient().createTopic(FACILITIES_TOPIC_NAME);
    getClient().subscribe(new SubscribeRequest(topicResult.getTopicArn(), EMAIL_PROTOCOL, email));
}

@Override
public CreateTopicResult createTopic(CreateTopicRequest createTopicRequest) throws AmazonServiceException, AmazonClientException {
    CreateTopicResult createTopicResult = new CreateTopicResult();
    createTopicResult.setTopicArn(DEFAULT_TOPIC_ARN);
    return createTopicResult;
}

public String createNotificationTopicIfNotExist(String topicName){
    CreateTopicRequest createTopicRequest = new CreateTopicRequest(topicName);
    CreateTopicResult createTopicResult = snsClient.createTopic(createTopicRequest);
    return createTopicResult.getTopicArn();
}

public CreateTopicResult createTopic(String name) {
    return sns.createTopic(name);
}

@Override
public void subscribeForAllFacilities(String email) {
    CreateTopicResult topicResult = getClient().createTopic(FACILITIES_TOPIC_NAME);
    getClient().subscribe(new SubscribeRequest(topicResult.getTopicArn(), EMAIL_PROTOCOL, email));
}

@Override
public Topic createTopic(String name) {
    return createTopic(name, (ResultCapture<CreateTopicResult>)null);
}

@Test
public void shouldSendNotificationMessageOnTopic() throws CfnAssistException, MissingArgumentException, InterruptedException, IOException {
    User user  = new User("path", "userName", "userId", "userArn", new Date());
    CFNAssistNotification notification = new CFNAssistNotification("name", "complete", user);

    CreateTopicResult createResult = snsClient.createTopic(SNSNotificationSender.TOPIC_NAME);
    String SNSarn = createResult.getTopicArn();
    assertNotNull(SNSarn);

    // test the SNS notification by creating a SQS and subscribing that to the SNS
    CreateQueueResult queueResult = createQueue();
    String queueUrl = queueResult.getQueueUrl();

    // give queue perms to subscribe to SNS
    Map<String, String> attribrutes = policyManager.getQueueAttributes(queueUrl);
    String queueArn = attribrutes.get(QueuePolicyManager.QUEUE_ARN_KEY);
    policyManager.checkOrCreateQueuePermissions(attribrutes, SNSarn, queueArn, queueUrl);

    // create subscription
    SubscribeRequest subscribeRequest = new SubscribeRequest(SNSarn, SNSEventSource.SQS_PROTO, queueArn);
    SubscribeResult subResult = snsClient.subscribe(subscribeRequest);
    String subscriptionArn = subResult.getSubscriptionArn();

    // send SNS and then check right thing arrives at SQS
    sender.sendNotification(notification);

    ReceiveMessageRequest request = new ReceiveMessageRequest().
            withQueueUrl(queueUrl).
            withWaitTimeSeconds(10);
    ReceiveMessageResult receiveResult = sqsClient.receiveMessage(request);
    List<Message> messages = receiveResult.getMessages();

    sqsClient.deleteQueue(queueUrl);
    snsClient.unsubscribe(subscriptionArn);
    snsClient.deleteTopic(SNSarn);

    assertEquals(1, messages.size());

    Message msg = messages.get(0);

    ObjectMapper objectMapper = new ObjectMapper();
    JsonNode rootNode = objectMapper.readTree(msg.getBody());       
    JsonNode messageNode = rootNode.get("Message");

    String json = messageNode.textValue();

    CFNAssistNotification result = CFNAssistNotification.fromJSON(json);

    assertEquals(notification, result);
}

public static String getTopicArn( final AmazonSNSClient sns, final String queueName, final boolean createOnMissing )
    throws Exception {

    if ( logger.isTraceEnabled() ) {
        logger.trace( "Looking up Topic ARN: {}", queueName );
    }

    ListTopicsResult listTopicsResult = sns.listTopics();
    String topicArn = null;

    for ( Topic topic : listTopicsResult.getTopics() ) {
        String arn = topic.getTopicArn();

        if ( queueName.equals( arn.substring( arn.lastIndexOf( ':' ) ) ) ) {
            topicArn = arn;

            if (logger.isTraceEnabled()) {
                logger.trace( "Found existing topic arn=[{}] for queue=[{}]", topicArn, queueName );
            }
        }
    }

    if ( topicArn == null && createOnMissing ) {
        if (logger.isTraceEnabled()) {
            logger.trace("Creating topic for queue=[{}]...", queueName);
        }

        CreateTopicResult createTopicResult = sns.createTopic( queueName );
        topicArn = createTopicResult.getTopicArn();

        if (logger.isTraceEnabled()) {
            logger.trace("Successfully created topic with name {} and arn {}", queueName, topicArn);
        }
    }
    else {
        logger.error( "Error looking up topic ARN for queue=[{}] and createOnMissing=[{}]", queueName,
            createOnMissing );
    }

    if ( logger.isTraceEnabled() ) {
        logger.trace( "Returning Topic ARN=[{}] for Queue=[{}]", topicArn, queueName );
    }


    return topicArn;
}

public void process(Exchange exchange) throws Exception {

        // TODO cache arns and don't create if not necessary
        final String topic = determineTopic(exchange);

        // creates a new topic, or returns the URL of an existing one
        CreateTopicRequest request = new CreateTopicRequest(topic);

        LOG.trace("Creating topic [{}] with request [{}]...", topic, request);

        final AmazonSNS snsClient = getEndpoint().getSNSClient();

        CreateTopicResult result = snsClient.createTopic(request);

        final String topicArn = result.getTopicArn();
        LOG.trace("Topic created with Amazon resource name: {}", topicArn);

        final SnsConfiguration configuration = getEndpoint().getConfiguration();

        if (ObjectHelper.isNotEmpty(configuration.getPolicy())) {
            LOG.trace("Updating topic [{}] with policy [{}]", topicArn, configuration.getPolicy());

            snsClient.setTopicAttributes(
                    new SetTopicAttributesRequest(topicArn, "Policy", configuration.getPolicy()));

            LOG.trace("Topic policy updated");
        }

        PublishRequest publishRequest = new PublishRequest();

        publishRequest.setTopicArn(topicArn);
        publishRequest.setSubject(determineSubject(exchange));
        publishRequest.setMessageStructure(determineMessageStructure(exchange));
        publishRequest.setMessage(exchange.getIn().getBody(String.class));

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

        PublishResult publishResult = snsClient.publish(publishRequest);

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

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

public void process(Exchange exchange) throws Exception {

        // TODO cache arns and don't create if not necessary
        final String topic = determineTopic(exchange);

        // creates a new topic, or returns the URL of an existing one
        CreateTopicRequest request = new CreateTopicRequest(topic);

        LOG.trace("Creating topic [{}] with request [{}]...", topic, request);

        final AmazonSNS snsClient = getEndpoint().getSNSClient();

        CreateTopicResult result = snsClient.createTopic(request);

        final String topicArn = result.getTopicArn();
        LOG.trace("Topic created with Amazon resource name: {}", topicArn);

        final SnsConfiguration configuration = getEndpoint().getConfiguration();

        if (ObjectHelper.isNotEmpty(configuration.getPolicy())) {
            LOG.trace("Updating topic [{}] with policy [{}]", topicArn, configuration.getPolicy());

            snsClient.setTopicAttributes(
                    new SetTopicAttributesRequest(topicArn, "Policy", configuration.getPolicy()));

            LOG.trace("Topic policy updated");
        }

        PublishRequest publishRequest = new PublishRequest();

        publishRequest.setTopicArn(topicArn);
        publishRequest.setSubject(determineSubject(exchange));
        publishRequest.setMessageStructure(determineMessageStructure(exchange));
        publishRequest.setMessage(exchange.getIn().getBody(String.class));

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

        PublishResult publishResult = snsClient.publish(publishRequest);

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

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

/**
 * Performs the <code>CreateTopic</code> action and use a ResultCapture to
 * retrieve the low-level client response.
 *
 * <p>
 *
 * @return The <code>Topic</code> resource object associated with the result
 *         of this action.
 * @see CreateTopicRequest
 */
com.amazonaws.resources.sns.Topic createTopic(CreateTopicRequest request,
        ResultCapture<CreateTopicResult> extractor);

/**
 * The convenient method form for the <code>CreateTopic</code> action.
 *
 * @see #createTopic(CreateTopicRequest, ResultCapture)
 */
com.amazonaws.resources.sns.Topic createTopic(String name,
        ResultCapture<CreateTopicResult> extractor);