Improvements to AWS topic subscription and SQS access policy generation

[AndriiLab]

There are few problems with DotNetCore.CAP.AmazonSQS provider. Our setup includes about 150 topics and about 15 queues. CAP is configured with ConsumerThreadCount = 4. This results in about 700 subscriptions.

Problem 1
When executing CAP with this configuration the subscription to SNS fails with the error AmazonSimpleNotificationServiceException: Rate exceeded and then required queues not created. This relates to the
snsClient.CreateTopicAsync call and occurs because AWS cannot process so many concurrent requests of SNS topic creation at a time.

Proposed solution to Problem 1
Separate topic creation and subscription process (i.e. extract topic creation out of the consumer threads). I have added FetchTopics method to IConsumerClient which returns a list of topics by default, but for AmazonSQS provider it handles the logic for SNS topic creation and/or its ARN fetching. This method is called before consumer threads creation.

Problem 2
When subscribing more than 20 SNS topics to the queue the access policy generated incorrectly and only the first 20 topics are able to post messages into the queue.

Proposed solution to Problem 2
This relates to the internal limit for SQS access policy to include only 20 statements. Thus to enable more than 20 SNS topics access to a single SQS we need to compact policy from:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "*"
      },
      "Action": "sqs:SendMessage",
      "Resource": "arn:aws:sqs:us-east-1:MyQueue",
      "Condition": {
        "ArnLike": {
          "aws:SourceArn": "arn:aws:sns:us-east-1:FirstTopic"
        }
      }
    },
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "*"
      },
      "Action": "sqs:SendMessage",
      "Resource": "arn:aws:sqs:us-east-1:MyQueue",
      "Condition": {
        "ArnLike": {
          "aws:SourceArn": "arn:aws:sns:us-east-1:SecondTopic"
        }
      }
    }]
}

to the compacted version with a single statement:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "*"
      },
      "Action": "sqs:SendMessage",
      "Resource": "arn:aws:sqs:us-east-1:MyQueue",
      "Condition": {
        "ArnLike": {
          "aws:SourceArn": [
          "arn:aws:sns:us-east-1:FirstTopic",
          "arn:aws:sns:us-east-1:SecondTopic"
          ]
        }
      }
    }]
}

After examination of AWS SDK, I have ended up with AmazonPolicyExtensions extensions, which help to achieve this.

Both improvements affect only DotNetCore.CAP.AmazonSQS provider, but require changes in DotNetCore.CAP since the execution of the FetchTopics method needs to be added to ConsumerRegister

[yang-xiaodong]

Looks good, Sorry for the delayed review

[yang-xiaodong]

Hello @AndriiLab ,

This PR included in version 5.0.1-preview-133783177 and released to nuget.

Do you have interest to join @dotnetcore/cap-committers of CAP as a maintainer?

[AndriiLab]

Hello @yang-xiaodong,
Thank you for accepting the suggested changes.
Also, thank you for the invitation!
I haven't very deep knowledge of the whole project, but got some understanding of how DotNetCore.CAP.AmazonSQS and core DotNetCore.CAP organized. If this is sound OK to you, I would be happy to help