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

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

@Override
public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) {
    log.info("Checking subscriptions to {}", config.aws.topic.subscribeArn);

    String thisEndpoint = "http://" + appInfo.getPublicHostname() + SnsCamEchoResponseEndpoint.SNS_EP_CAM_ECHO_RESPONSE;

    List<Subscription> subscriptions = amazonSns.listSubscriptionsByTopic(config.aws.topic.subscribeArn).getSubscriptions();
    boolean notSubscribed = true;
    for (Subscription subscription : subscriptions) {
        if (thisEndpoint.equals(subscription.getEndpoint())) {
            log.info("Found subscription {} on topic {} for endpoint {}", subscription.getSubscriptionArn(), subscription.getTopicArn(), thisEndpoint);
            notSubscribed = false;
        }
    }

    if (notSubscribed) {
        log.info("Subscribing to topic {} with endpoint {}", config.aws.topic.subscribeArn, thisEndpoint);
        SubscribeResult subscribeResult = amazonSns.subscribe(config.aws.topic.subscribeArn, "http", thisEndpoint);
        log.info("Subscription: {}", subscribeResult.getSubscriptionArn());
    }
}

public String subscribeQueueToTopic(String snsTopicArn, String sqsQueueUrl){        
       Map<String, String> queueAttributes = sqsClient.getQueueAttributes(new GetQueueAttributesRequest(sqsQueueUrl)
               .withAttributeNames(QueueAttributeName.QueueArn.toString())).getAttributes();
       String sqsQueueArn = queueAttributes.get(QueueAttributeName.QueueArn.toString());

       Policy policy = new Policy().withStatements(
               new Statement(Effect.Allow)
                   .withId("topic-subscription-" + snsTopicArn)
                   .withPrincipals(Principal.AllUsers)
                   .withActions(SQSActions.SendMessage)
                   .withResources(new Resource(sqsQueueArn))
                   .withConditions(ConditionFactory.newSourceArnCondition(snsTopicArn)));

       logger.debug("Policy: " + policy.toJson());

       queueAttributes = new HashMap<String, String>();
       queueAttributes.put(QueueAttributeName.Policy.toString(), policy.toJson());
       sqsClient.setQueueAttributes(new SetQueueAttributesRequest(sqsQueueUrl, queueAttributes));

       SubscribeResult subscribeResult =
               snsClient.subscribe(new SubscribeRequest()
                   .withEndpoint(sqsQueueArn)
                   .withProtocol("sqs")
                   .withTopicArn(snsTopicArn));
       return subscribeResult.getSubscriptionArn();
}

@Override
public SubscribeResult subscribe(SubscribeRequest subscribeRequest) throws AmazonClientException {
    final String protocol = subscribeRequest.getProtocol().toLowerCase();
    if (!protocol.equals("sqs")) {
        throw new InvalidParameterException("endpoint protocol " + protocol + " not supported");
    }
    final String topicArn = subscribeRequest.getTopicArn();
    if (!_subscriptionsForTopic.containsKey(topicArn)) {
        throw new InvalidParameterException("no such topic " + topicArn);
    }
    String subscriptionArn = topicArn + ":" + RandomStringUtils.randomNumeric(7);
    if (!_subscriptionsByArn.containsKey(subscriptionArn)) {
        _subscriptionsByArn.put(subscriptionArn, new Subscription().
                withTopicArn(topicArn).
                withProtocol(protocol).
                withSubscriptionArn(subscriptionArn).
                withEndpoint(subscribeRequest.getEndpoint()));
        _subscriptionsForTopic.get(topicArn).add(subscriptionArn);
    }

    return new SubscribeResult().withSubscriptionArn(subscriptionArn);
}

@Override
public Subscription subscribe(SubscribeRequest request,
        ResultCapture<SubscribeResult> extractor) {

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

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

@Override
public Subscription subscribe(String endpoint, String protocol,
        ResultCapture<SubscribeResult> extractor) {

    SubscribeRequest request = new SubscribeRequest()
        .withEndpoint(endpoint)
        .withProtocol(protocol);
    return subscribe(request, extractor);
}

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

private void processUrlSubscription(Subscription urlSubscription) {

        String snsUrlSubscriptionArn = null;
        for (Subscription subscription : client.listSubscriptions()
                .getSubscriptions()) {
            if (subscription.getTopicArn().equals(topicArn)
                    && subscription.getProtocol().equals(
                            urlSubscription.getProtocol())
                    && subscription.getEndpoint().contains(
                            urlSubscription.getEndpoint())) {
                if (!subscription.getSubscriptionArn().equals(
                        "PendingConfirmation")) {
                    snsUrlSubscriptionArn = subscription.getSubscriptionArn();
                    break;
                }
            }
        }
        if (snsUrlSubscriptionArn == null) {

            SubscribeRequest request = new SubscribeRequest(topicArn,
                    urlSubscription.getProtocol(),
                    urlSubscription.getEndpoint());
            SubscribeResult result = client.subscribe(request);
            snsUrlSubscriptionArn = result.getSubscriptionArn();
            log.info("Subscribed URL to SNS with subscription ARN: "
                    + snsUrlSubscriptionArn);
        } else {
            log.info("Already subscribed with ARN: " + snsUrlSubscriptionArn);
        }
    }

@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 SubscribeResult subscribeToTopic(String arn,
        String protocol, String endpoint) {
    SubscribeRequest subscribe = new SubscribeRequest(arn, protocol,
            endpoint);
    return snsClient.subscribe(subscribe);
}

@Override
public SubscribeResult subscribe(SubscribeRequest subscribeRequest) throws AmazonServiceException, AmazonClientException {
    throw new UnsupportedOperationException();
}

@Override
public Subscription subscribe(String endpoint, String protocol) {
    return subscribe(endpoint, protocol,
            (ResultCapture<SubscribeResult>)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);
}

private void processSqsSubscription(Subscription sqsSubscription) {
    Assert.state(sqsExecutorMap != null,
            "'sqsExecutorMap' must not be null");

    SqsExecutor sqsExecutor = null;
    String endpointValue = sqsSubscription.getEndpoint();
    if (sqsExecutorMap.containsKey(endpointValue)) {
        sqsExecutor = sqsExecutorMap.get(endpointValue);
        sqsSubscription.setEndpoint(sqsExecutor.getQueueArn());
    } else {
        // endpointValue is the queue-arn
        sqsSubscription.setEndpoint(endpointValue);
    }

    String snsSqsSubscriptionArn = null;
    for (Subscription subscription : client.listSubscriptions()
            .getSubscriptions()) {
        if (subscription.getTopicArn().equals(topicArn)
                && subscription.getProtocol().equals(
                        sqsSubscription.getProtocol())
                && subscription.getEndpoint().equals(
                        sqsSubscription.getEndpoint())) {
            snsSqsSubscriptionArn = subscription.getSubscriptionArn();
            break;
        }
    }
    if (snsSqsSubscriptionArn == null) {
        SubscribeRequest request = new SubscribeRequest(topicArn,
                sqsSubscription.getProtocol(),
                sqsSubscription.getEndpoint());
        SubscribeResult result = client.subscribe(request);
        snsSqsSubscriptionArn = result.getSubscriptionArn();
        log.info("Subscribed SQS to SNS with subscription ARN: "
                + snsSqsSubscriptionArn);
    } else {
        log.info("Already subscribed with ARN: " + snsSqsSubscriptionArn);
    }
    if (sqsExecutor != null) {
        sqsExecutor.addSnsPublishPolicy(topicName, topicArn);
    }
}

/**
 * Performs the <code>Subscribe</code> action and use a ResultCapture to
 * retrieve the low-level client response.
 *
 * <p>
 * The following request parameters will be populated from the data of this
 * <code>Topic</code> resource, and any conflicting parameter value set in
 * the request will be overridden:
 * <ul>
 *   <li>
 *     <b><code>TopicArn</code></b>
 *         - mapped from the <code>Arn</code> identifier.
 *   </li>
 * </ul>
 *
 * <p>
 *
 * @return The <code>Subscription</code> resource object associated with the
 *         result of this action.
 * @see SubscribeRequest
 */
com.amazonaws.resources.sns.Subscription subscribe(SubscribeRequest request,
        ResultCapture<SubscribeResult> extractor);

/**
 * The convenient method form for the <code>Subscribe</code> action.
 *
 * @see #subscribe(SubscribeRequest, ResultCapture)
 */
com.amazonaws.resources.sns.Subscription subscribe(String endpoint, String
        protocol, ResultCapture<SubscribeResult> extractor);