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

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

private void setQueuePolicy(String topicSnsArn, String queueArn, String queueURL) {
    logger.info("Set up policy for queue to allow SNS to publish to it");
    Policy sqsPolicy = new Policy()
       .withStatements(new Statement(Statement.Effect.Allow)
                           .withPrincipals(Principal.AllUsers)
                           .withResources(new Resource(queueArn))
                           .withConditions(ConditionFactory.newSourceArnCondition(topicSnsArn))
                           .withActions(SQSActions.SendMessage));

       Map<String, String> attributes = new HashMap<String,String>();
       attributes.put("Policy", sqsPolicy.toJson());

       SetQueueAttributesRequest setQueueAttributesRequest = new SetQueueAttributesRequest();
       setQueueAttributesRequest.setQueueUrl(queueURL);
       setQueueAttributesRequest.setAttributes(attributes);
       sqsClient.setQueueAttributes(setQueueAttributesRequest);     
}

@Test
public void testNonInjectableMocks_shouldReturnNormal() {
  assertNotNull(sqs.changeMessageVisibilityBatch(new ChangeMessageVisibilityBatchRequest()));
  assertNotNull(sqs.addPermission(new AddPermissionRequest().withActions("one").withAWSAccountIds("two","three").withLabel("four").withQueueUrl("five")));
  assertNotNull(sqs.listDeadLetterSourceQueues(new ListDeadLetterSourceQueuesRequest().withQueueUrl("ten")));
  assertNotNull(sqs.getQueueAttributes(new GetQueueAttributesRequest().withAttributeNames(ImmutableList.of("eleven")).withQueueUrl("twelve")));
  assertNotNull(sqs.setQueueAttributes(new SetQueueAttributesRequest().withAttributes(ImmutableMap.of("thirteen","fourteen")).withQueueUrl("fifteen")));
}

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

@Override
public SetQueueAttributesResult setQueueAttributes(SetQueueAttributesRequest setQueueAttributesRequest) throws AmazonServiceException, AmazonClientException {
    synchronized (queueAttributes) {
        if (!queueAttributes.containsKey(setQueueAttributesRequest.getQueueUrl())) {
            queueAttributes.put(setQueueAttributesRequest.getQueueUrl(), new HashMap<String, String>());
        }
        for (final Map.Entry<String, String> entry : setQueueAttributesRequest.getAttributes().entrySet()) {
            queueAttributes.get(setQueueAttributesRequest.getQueueUrl()).put(entry.getKey(), entry.getValue());
        }
    }
    return new SetQueueAttributesResult();
}

@Override
public void setAttributes(Map<String, String> attributes,
        ResultCapture<Void> extractor) {

    SetQueueAttributesRequest request = new SetQueueAttributesRequest()
        .withAttributes(attributes);
    setAttributes(request, extractor);
}

@VisibleForTesting
static SetQueueAttributesRequest createRequest(String queueUrl, MutableSqsQueueAttributes attributes) {
    return new SetQueueAttributesRequest()
            .withQueueUrl(queueUrl)
            .withAttributes(attributes.getStringMap());
}

@Test
public void testCreateRequest() {
    SetQueueAttributesRequest request = SetQueueAttributesAction.createRequest(QUEUE_URL, ATTRIBUTES);
    assertThat(request.getQueueUrl()).isEqualTo(QUEUE_URL);
    assertThat(request.getAttributes()).isEmpty();
}

public Observable<SetQueueAttributesResult> setQueueAttributesAsync(SetQueueAttributesRequest request) {
    return Observable.from(sqsClient.setQueueAttributesAsync(request));
}

public static void main(String[] args)
{
    final String USAGE =
       "To run this example, supply the name of a queue to create and\n" +
       "queue url of an existing queue.\n\n" +
       "Ex: LongPolling <unique-queue-name> <existing-queue-url>\n";

    if (args.length != 2) {
        System.out.println(USAGE);
        System.exit(1);
    }

    String queue_name = args[0];
    String queue_url = args[1];

    final AmazonSQS sqs = AmazonSQSClientBuilder.defaultClient();

    // Enable long polling when creating a queue
    CreateQueueRequest create_request = new CreateQueueRequest()
            .withQueueName(queue_name)
            .addAttributesEntry("ReceiveMessageWaitTimeSeconds", "20");

    try {
        sqs.createQueue(create_request);
    } catch (AmazonSQSException e) {
        if (!e.getErrorCode().equals("QueueAlreadyExists")) {
            throw e;
        }
    }

    // Enable long polling on an existing queue
    SetQueueAttributesRequest set_attrs_request = new SetQueueAttributesRequest()
            .withQueueUrl(queue_url)
            .addAttributesEntry("ReceiveMessageWaitTimeSeconds", "20");
    sqs.setQueueAttributes(set_attrs_request);

    // Enable long polling on a message receipt
    ReceiveMessageRequest receive_request = new ReceiveMessageRequest()
            .withQueueUrl(queue_url)
            .withWaitTimeSeconds(20);
    sqs.receiveMessage(receive_request);
}

@Override
public SetQueueAttributesResult setQueueAttributes(SetQueueAttributesRequest setQueueAttributesRequest) throws AmazonServiceException, AmazonClientException {
    return new SetQueueAttributesResult();
}

@Override
public void setAttributes(SetQueueAttributesRequest request) {
    setAttributes(request, null);
}

@Override
public void setAttributes(SetQueueAttributesRequest request,
        ResultCapture<Void> extractor) {

    resource.performAction("SetAttributes", request, extractor);
}

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

/**
 * Performs the <code>SetAttributes</code> action.
 *
 * <p>
 * The following request parameters will be populated from the data of this
 * <code>Queue</code> resource, and any conflicting parameter value set in
 * the request will be overridden:
 * <ul>
 *   <li>
 *     <b><code>QueueUrl</code></b>
 *         - mapped from the <code>Url</code> identifier.
 *   </li>
 * </ul>
 *
 * <p>
 *
 * @see SetQueueAttributesRequest
 */
void setAttributes(SetQueueAttributesRequest request);

/**
 * Performs the <code>SetAttributes</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>Queue</code> resource, and any conflicting parameter value set in
 * the request will be overridden:
 * <ul>
 *   <li>
 *     <b><code>QueueUrl</code></b>
 *         - mapped from the <code>Url</code> identifier.
 *   </li>
 * </ul>
 *
 * <p>
 *
 * @see SetQueueAttributesRequest
 */
void setAttributes(SetQueueAttributesRequest request, ResultCapture<Void>
        extractor);

/**
 * <p>
 * Sets the value of one or more queue attributes. When you change a queue's
 * attributes, the change can take up to 60 seconds for most of the
 * attributes to propagate throughout the SQS system. Changes made to the
 * <code>MessageRetentionPeriod</code> attribute can take up to 15 minutes.
 * </p>
 * <p>
 * <b>NOTE:</b>Going forward, new attributes might be added. If you are
 * writing code that calls this action, we recommend that you structure your
 * code so that it can handle new attributes gracefully.
 * </p>
 *
 * @param setQueueAttributesRequest
 *            Container for the necessary parameters to execute the
 *            SetQueueAttributes service method on AmazonSQS.
 * 
 * 
 * @throws InvalidAttributeNameException
 *
 * @throws AmazonClientException
 *             If any internal errors are encountered inside the client
 *             while attempting to make the request or handle the response.
 *             For example if a network connection is not available.
 * @throws AmazonServiceException
 *             If an error response is returned by AmazonSQS indicating
 *             either a problem with the data in the request, or a server
 *             side issue.
 */
public SetQueueAttributesResult setQueueAttributes(SetQueueAttributesRequest setQueueAttributesRequest)
        throws AmazonServiceException, AmazonClientException {

    return amazonSqsToBeExtended.setQueueAttributes(setQueueAttributesRequest);

}