Java 类com.amazonaws.services.sqs.model.QueueAttributeName 实例源码

@Derived
public Map<String, String> getStringMap() {
    ImmutableMap.Builder<String, String> builder = ImmutableMap.<String, String>builder();
    getDeliveryDelay().ifPresent(value -> {
        builder.put(QueueAttributeName.DelaySeconds.toString(), Long.toString(value.getSeconds()));
    });
    getVisibilityTimeout().ifPresent(value -> {
        builder.put(QueueAttributeName.VisibilityTimeout.toString(), Long.toString(value.getSeconds()));
    });
    getMaxMessageBytes().ifPresent(value -> {
        builder.put(QueueAttributeName.MaximumMessageSize.toString(), Integer.toString(value));
    });
    getMessageRetentionPeriod().ifPresent(value -> {
        builder.put(QueueAttributeName.MessageRetentionPeriod.toString(), Long.toString(value.getSeconds()));
    });
    getRedrivePolicy().ifPresent((value -> {
        builder.put(QueueAttributeName.RedrivePolicy.toString(), value.toAttributeString());
    }));
    return builder.build();
}

@Test
public void testShouldReturnTotalNumberOfFailedEvents() {

    final SdkHttpMetadata responseMetadata = mock(SdkHttpMetadata.class);
    when(responseMetadata.getHttpStatusCode()).thenReturn(200);

    final String totalNumberOfFailedEvents = RandomStringUtils.randomNumeric(4);
    final Map<String, String> attributes = new HashMap<>();
    attributes.put(QueueAttributeName.ApproximateNumberOfMessages.name(), totalNumberOfFailedEvents);

    final GetQueueAttributesResult getQueueAttributesResult = new GetQueueAttributesResult();
    getQueueAttributesResult.setSdkHttpMetadata(responseMetadata);
    getQueueAttributesResult.setAttributes(attributes);

    when(amazonSQS.getQueueAttributes(any(GetQueueAttributesRequest.class))).thenReturn(getQueueAttributesResult);

    assertThat(sqsFailedEventSource.getSize()).isEqualTo(Long.valueOf(totalNumberOfFailedEvents));
}

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

public ImmutableSqsQueueAttributes.Builder fromStringMap(Map<String, String> map) {
    long delaySeconds = Long.parseLong(map.get(QueueAttributeName.DelaySeconds.toString()));
    long visibilitySeconds = Long.parseLong(map.get(QueueAttributeName.VisibilityTimeout.toString()));
    int messageBytes = Integer.parseInt(map.get(QueueAttributeName.MaximumMessageSize.toString()));
    long messageRetentionSeconds =
            Long.parseLong(map.get(QueueAttributeName.MessageRetentionPeriod.toString()));
    String queueArn = map.get(QueueAttributeName.QueueArn.toString());

    return builder()
            .deliveryDelay(Duration.ofSeconds(delaySeconds))
            .visibilityTimeout(Duration.ofSeconds(visibilitySeconds))
            .maxMessageBytes(messageBytes)
            .messageRetentionPeriod(Duration.ofSeconds(messageRetentionSeconds))
            .queueArn(queueArn);
}

@Override
public long getSize() {

    final GetQueueAttributesRequest getQueueAttributesRequest = new GetQueueAttributesRequest(queueUrl,
            Collections.singletonList(QueueAttributeName.ApproximateNumberOfMessages.name()));

    final GetQueueAttributesResult queueAttributes = amazonSQS.getQueueAttributes(getQueueAttributesRequest);

    if (queueAttributes.getAttributes() != null) {
        return Long.valueOf(queueAttributes.getAttributes().getOrDefault(
                    QueueAttributeName.ApproximateNumberOfMessages.name(), "0"));
    } else {
        return 0L;
    }
}

protected void createQueue(AmazonSQS client) {
    LOG.trace("Queue '{}' doesn't exist. Will create it...", configuration.getQueueName());

    // creates a new queue, or returns the URL of an existing one
    CreateQueueRequest request = new CreateQueueRequest(configuration.getQueueName());
    if (getConfiguration().getDefaultVisibilityTimeout() != null) {
        request.getAttributes().put(QueueAttributeName.VisibilityTimeout.name(), String.valueOf(getConfiguration().getDefaultVisibilityTimeout()));
    }
    if (getConfiguration().getMaximumMessageSize() != null) {
        request.getAttributes().put(QueueAttributeName.MaximumMessageSize.name(), String.valueOf(getConfiguration().getMaximumMessageSize()));
    }
    if (getConfiguration().getMessageRetentionPeriod() != null) {
        request.getAttributes().put(QueueAttributeName.MessageRetentionPeriod.name(), String.valueOf(getConfiguration().getMessageRetentionPeriod()));
    }
    if (getConfiguration().getPolicy() != null) {
        request.getAttributes().put(QueueAttributeName.Policy.name(), String.valueOf(getConfiguration().getPolicy()));
    }
    if (getConfiguration().getReceiveMessageWaitTimeSeconds() != null) {
        request.getAttributes().put(QueueAttributeName.ReceiveMessageWaitTimeSeconds.name(), String.valueOf(getConfiguration().getReceiveMessageWaitTimeSeconds()));
    }
    if (getConfiguration().getRedrivePolicy() != null) {
        request.getAttributes().put(QueueAttributeName.RedrivePolicy.name(), getConfiguration().getRedrivePolicy());
    }
    LOG.trace("Creating queue [{}] with request [{}]...", configuration.getQueueName(), request);

    CreateQueueResult queueResult = client.createQueue(request);
    queueUrl = queueResult.getQueueUrl();

    LOG.trace("Queue created and available at: {}", queueUrl);
}

private void updateQueueAttributes(AmazonSQS client) {
    SetQueueAttributesRequest request = new SetQueueAttributesRequest();
    request.setQueueUrl(queueUrl);
    if (getConfiguration().getDefaultVisibilityTimeout() != null) {
        request.getAttributes().put(QueueAttributeName.VisibilityTimeout.name(), String.valueOf(getConfiguration().getDefaultVisibilityTimeout()));
    }
    if (getConfiguration().getMaximumMessageSize() != null) {
        request.getAttributes().put(QueueAttributeName.MaximumMessageSize.name(), String.valueOf(getConfiguration().getMaximumMessageSize()));
    }
    if (getConfiguration().getMessageRetentionPeriod() != null) {
        request.getAttributes().put(QueueAttributeName.MessageRetentionPeriod.name(), String.valueOf(getConfiguration().getMessageRetentionPeriod()));
    }
    if (getConfiguration().getPolicy() != null) {
        request.getAttributes().put(QueueAttributeName.Policy.name(), String.valueOf(getConfiguration().getPolicy()));
    }
    if (getConfiguration().getReceiveMessageWaitTimeSeconds() != null) {
        request.getAttributes().put(QueueAttributeName.ReceiveMessageWaitTimeSeconds.name(), String.valueOf(getConfiguration().getReceiveMessageWaitTimeSeconds()));
    }
    if (getConfiguration().getRedrivePolicy() != null) {
        request.getAttributes().put(QueueAttributeName.RedrivePolicy.name(), getConfiguration().getRedrivePolicy());
    }
    if (!request.getAttributes().isEmpty()) {
        LOG.trace("Updating queue '{}' with the provided queue attributes...", configuration.getQueueName());
        client.setQueueAttributes(request);
        LOG.trace("Queue '{}' updated and available at {}'", configuration.getQueueName(), queueUrl);
    }
}

private static String getSQSQueueARN(AmazonSQS amazonSQS, String queueURL) {
    // This statement will throw if the queue does not exist.
    GetQueueAttributesResult queueAttributes = amazonSQS.getQueueAttributes(
        new GetQueueAttributesRequest()
            .withQueueUrl(queueURL)
            .withAttributeNames(QueueAttributeName.QueueArn)
    );
    return queueAttributes
        .getAttributes()
        .get(QueueAttributeName.QueueArn.name());
}

private static void mockGetQueueAttributesWithRedrivePolicy(AmazonSQSAsync sqs, String queueUrl) {
    when(sqs.getQueueAttributes(new GetQueueAttributesRequest(queueUrl).withAttributeNames(QueueAttributeName.RedrivePolicy))).
            thenReturn(new GetQueueAttributesResult().addAttributesEntry(QueueAttributeName.RedrivePolicy.toString(), "{\"some\": \"JSON\"}"));
}

private static void mockGetQueueAttributesWithEmptyResult(AmazonSQSAsync sqs, String queueUrl) {
    when(sqs.getQueueAttributes(new GetQueueAttributesRequest(queueUrl).withAttributeNames(QueueAttributeName.RedrivePolicy))).
            thenReturn(new GetQueueAttributesResult());
}

private static void allowSQSQueueToReceiveMessagesFromSNSTopic(
    AmazonSQS amazonSQS,
    String queueURL,
    String queueARN,
    String topicARN
) {
    GetQueueAttributesResult queueAttributesResult =
        amazonSQS.getQueueAttributes(
            new GetQueueAttributesRequest().withQueueUrl(queueURL).withAttributeNames(
                QueueAttributeName.Policy
            )
        );

    String policyJson = queueAttributesResult.getAttributes().get(QueueAttributeName.Policy.name());

    final List<Statement> statements;
    if (policyJson != null) {
        statements = new ArrayList<>(Policy.fromJson(policyJson).getStatements());
    } else {
        // no policies yet exist
        statements = new ArrayList<>();
    }

    statements.add(
        new Statement(Statement.Effect.Allow)
            .withPrincipals(Principal.AllUsers)
            .withResources(new Resource(queueARN))
            .withActions(SQSActions.SendMessage)
            .withConditions(ConditionFactory.newSourceArnCondition(topicARN))
    );

    Policy policy = new Policy();
    policy.setStatements(statements);
    Map<String, String> queueAttributes = new HashMap<>();
    queueAttributes.put(QueueAttributeName.Policy.name(), policy.toJson());

    // Note that if the queue already has this policy, this will do nothing.
    amazonSQS.setQueueAttributes(
        new SetQueueAttributesRequest()
            .withQueueUrl(queueURL)
            .withAttributes(queueAttributes)
    );
}