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

@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 run(CreateCloudFrontSecurityGroupUpdaterLambdaCommand command) {
    if (! cloudFormationService.isStackPresent(StackName.CLOUD_FRONT_IP_SYNCHRONIZER.getName())) {
        createLambda();
    }

    Map<String, String> outputs = cloudFormationService.getStackOutputs(StackName.CLOUD_FRONT_IP_SYNCHRONIZER.getName());
    CloudFrontIpSynchronizerOutputs cloudFrontIpSynchronizerOutputs = cloudformationObjectMapper
            .convertValue(outputs, CloudFrontIpSynchronizerOutputs.class);

    final String arn = cloudFrontIpSynchronizerOutputs.getCloudFrontOriginElbSgIpSyncFunctionArn();

    // subscribe, if already subscribed it doesn't make a new sub
    amazonSNS.subscribe(new SubscribeRequest(AWS_IP_CHANGE_TOPIC_ARN, "lambda", arn));

    // force any new ELBs that have the tags we care about to be updated to only allow ingress from CloudFront
    forceLambdaToUpdateSgs(arn);
}

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

/**
 * 
 * @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 shouldSubscribe_withSqsQueueResource() {
    // Given
    final SqsQueueResource mockSqsQueueResource = mock(SqsQueueResource.class);
    final String queueArn = randomString();
    when(mockSqsQueueResource.queueArn()).thenReturn(queueArn);

    // When
    snsTopicResource.subscribe(mockSqsQueueResource);

    // Then
    final ArgumentCaptor<SubscribeRequest> captor = ArgumentCaptor.forClass(SubscribeRequest.class);
    verify(mockAmazonSnsClient).subscribe(captor.capture());
    final SubscribeRequest subscribeRequest = captor.getValue();
    assertEquals(topicArn, subscribeRequest.getTopicArn());
    assertEquals("sqs", subscribeRequest.getProtocol());
    assertEquals(queueArn, subscribeRequest.getEndpoint());
}

@Test
public void shouldThrowException_onAmazonClientExceptionFromSubscribe() {
    // Given
    final SqsQueueResource mockSqsQueueResource = mock(SqsQueueResource.class);
    final String queueArn = randomString();
    when(mockSqsQueueResource.queueArn()).thenReturn(queueArn);
    when(mockAmazonSnsClient.subscribe(any(SubscribeRequest.class))).thenThrow(AmazonClientException.class);

    // When
    AmazonClientException thrownException = null;
    try {
        snsTopicResource.subscribe(mockSqsQueueResource);
    } catch (final AmazonClientException e) {
        thrownException = e;
    }

    // Then
    assertNotNull(thrownException);
}

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

private static void subscribeSQSQueueToSNSTopic(
    AmazonSNS amazonSNS,
    String queueARN,
    String topicARN
) {
    // Note that if there is already a subscription with these parameters, this will do nothing.
    amazonSNS.subscribe(
        new SubscribeRequest()
            .withTopicArn(topicARN)
            .withProtocol("sqs")
            .withEndpoint(queueARN)
    );
}

public void manualSubscriptionSetup() {
    final String queueName = someQueueName();
    final String queueUrl = someNewQueue(queueName);
    final String topicName = "manualSubscriptionSetup";
    final String message = "hi from " + topicName;

    AmazonSNS amazonSNS = new InMemorySNS(_amazonSQS1);

    amazonSNS.createTopic(new CreateTopicRequest(topicName));
    Assert.assertEquals(amazonSNS.listTopics().getTopics().size(), 1);
    Assert.assertEquals(amazonSNS.listTopics().getTopics().get(0).getTopicArn(), makeTopicArn(topicName));

    //make sure create is idempotent
    amazonSNS.createTopic(new CreateTopicRequest(topicName));
    Assert.assertEquals(amazonSNS.listTopics().getTopics().size(), 1, "existing topic duplicated?");
    Assert.assertEquals(amazonSNS.listTopics().getTopics().get(0).getTopicArn(), makeTopicArn(topicName));

    //subscribe.
    amazonSNS.subscribe(new SubscribeRequest().
            withEndpoint(getQueueArn(queueName)).
            withProtocol("sqs").
            withTopicArn(makeTopicArn(topicName)));

    Assert.assertEquals(amazonSNS.listSubscriptions().getSubscriptions().size(), 1);
    Assert.assertEquals(amazonSNS.listSubscriptions().getSubscriptions().get(0).getTopicArn(), makeTopicArn(topicName));
    Assert.assertEquals(amazonSNS.
            listSubscriptionsByTopic(new ListSubscriptionsByTopicRequest(makeTopicArn(topicName)))
            .getSubscriptions()
            .size(),
            1);

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

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

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

@Test
public void testSubscribe_withNoRequestParams_shouldWork() {
  mockSns(new MockParameters());
  assertNotNull(sns.subscribe(new SubscribeRequest()));
}

private SubscribeResult subscribeToTopic(String arn,
        String protocol, String endpoint) {
    SubscribeRequest subscribe = new SubscribeRequest(arn, protocol,
            endpoint);
    return snsClient.subscribe(subscribe);
}

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

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

public String subscribeEmailToTopic(String topicArn, String emailAddress){
    SubscribeRequest subRequest = new SubscribeRequest(topicArn, "email", emailAddress);
    return snsClient.subscribe(subRequest).getSubscriptionArn();
}

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

@Override
public Subscription subscribe(SubscribeRequest request) {
    return subscribe(request, 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);
    }
}

/**
 * Adds an AWS SQS subscription to the AWS SNS topic.
 * @param sqsQueueResource {@link SqsQueueResource} representative of AWS SQS queue subscribing to the AWS SNS
 *            topic.
 * @throws AmazonClientException
 */
public void subscribe(final SqsQueueResource sqsQueueResource) throws AmazonClientException {
    amazonSnsClient.subscribe(new SubscribeRequest(topicArn, "sqs", sqsQueueResource.queueArn()));
}

/**
 * Performs the <code>Subscribe</code> action.
 *
 * <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)
        ;

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