5

Here is what I tried, using version 53eb8b070e9a5067829fd029539966181632032a of aws-sdk-go.

// main.go
package main

import (
    "errors"
    "fmt"
    "log"
    "net/http"

    "github.com/aws/aws-sdk-go/aws"
    "github.com/aws/aws-sdk-go/aws/session"
    "github.com/aws/aws-sdk-go/service/sns"
    "github.com/aws/aws-sdk-go/service/sqs"
)

func main() {
    if err := makeTopicAndQueue(); err != nil {
        log.Fatalf("aws-sns-sqs: %v", err)
    }
}

func makeTopicAndQueue() error {
    sess, err := session.NewSession(&aws.Config{
        HTTPClient:  &http.Client{},
        Region:      aws.String("us-east-2"),
        Credentials: nil,
        MaxRetries:  aws.Int(0),
    })

    log.Printf("Creating an SNS topic.")
    snsClient := sns.New(sess, &aws.Config{})
    topicName := "test-topic"
    out, err := snsClient.CreateTopic(&sns.CreateTopicInput{Name: aws.String(topicName)})
    if err != nil {
        return fmt.Errorf(`creating topic "%s": %v`, topicName, err)
    }
    defer snsClient.DeleteTopic(&sns.DeleteTopicInput{TopicArn: out.TopicArn})

    log.Printf("Creating an SQS queue.")
    sqsClient := sqs.New(sess, &aws.Config{})
    subName := "test-subscription"
    out2, err := sqsClient.CreateQueue(&sqs.CreateQueueInput{QueueName: aws.String(subName)})
    if err != nil {
        return fmt.Errorf(`creating subscription queue "%s": %v`, subName, err)
    }

    log.Printf("Getting queue ARN.")
    out3, err := sqsClient.GetQueueAttributes(&sqs.GetQueueAttributesInput{
        QueueUrl:       out2.QueueUrl,
        AttributeNames: []*string{aws.String("QueueArn")},
    })
    if err != nil {
        return fmt.Errorf("getting queue ARN for %s: %v", *out2.QueueUrl, err)
    }
    qARN := out3.Attributes["QueueArn"]

    log.Printf("Subscribing the queue to the topic.")
    _, err = snsClient.Subscribe(&sns.SubscribeInput{
        TopicArn: out.TopicArn,
        Endpoint: qARN,
        Protocol: aws.String("sqs"),
    })
    if err != nil {
        return fmt.Errorf("subscribing: %v", err)
    }

    log.Printf("Getting the confirmation token from the queue.")
    out4, err := sqsClient.ReceiveMessage(&sqs.ReceiveMessageInput{
        QueueUrl: out2.QueueUrl,
    })
    if err != nil {
        return fmt.Errorf("receiving subscription confirmation message from queue: %v", err)
    }
    ms := out4.Messages
    var token *string
    switch len(ms) {
    case 0:
        return errors.New("no subscription confirmation message found in queue")
    case 1:
        m := ms[0]
        token = m.Body
    default:
        return fmt.Errorf("%d messages found in queue, want exactly 1", len(ms))
    }

    log.Printf("Using the token to finish subscribing.")
    _, err = snsClient.ConfirmSubscription(&sns.ConfirmSubscriptionInput{
        TopicArn: out.TopicArn,
        Token:    token,
    })
    if err != nil {
        return fmt.Errorf("confirming subscription: %v", err)
    }
    sqsClient.DeleteQueue(&sqs.DeleteQueueInput{QueueUrl: out2.QueueUrl})

    return nil
}

I expected it to get to the end but it failed with this output:

[ ~/src/aws-sqs-issue ] go run main.go
2019/01/15 09:31:19 Creating an SNS topic.
2019/01/15 09:31:19 Creating an SQS queue.
2019/01/15 09:31:20 Getting queue ARN.
2019/01/15 09:31:20 Subscribing the queue to the topic.
2019/01/15 09:31:21 Getting the confirmation token from the queue.
2019/01/15 09:31:21 aws-sns-sqs: no subscription confirmation message found in queue

Am I doing something wrong or is this a bug in the SDK?

I'm not sure what else to say about this. Here is some additional verbiage to somehow get the warning about the post being mostly code to go away. It's best to stop reading at this point because all the rest of this is bound to make for boring reading. I don't know much much longer I can keep on making up nonsense to satisfy this silly algorithm. Why don't they allow a simple post that contains a lot of code? I have no idea. Ah well. There is an upside down gopher on my desk. I think it was intentional. Due to the poor critter's anatomy, he's doing more of an eyeball-stand than a head-stand. My desk plant didn't hold up too well over the holidays. Better give it some water. Wowee, this thing really wants an awful lot of words. Well, I aim to please. If I keep going, will I accidentally output some Shakespeare? Well, it's over, thank the Bard.

2 Answers 2

4

This one is quite old, but here is the trick to solve it:

For an Amazon SNS topic to be able to send messages to a queue, you must set a policy on the queue that allows the Amazon SNS topic to perform the sqs:SendMessage action. See more

To do so using the current version of the V1 SDK, when creating the Queue you must define the Policy manually as a Queue attribute, allowing the SNS Topic to send messages to your SQS Queue.

Here is an example as code:

    queueARN := fmt.Sprintf("arn:aws:sqs:%s:%s:%s", "us-east-1", "xxx", "my-queue")
    topicARN := fmt.Sprintf("arn:aws:sns:%s:%s:%s", "us-east-1", "xxx", "my-topic")
    _, err := b.sqs.CreateQueue(&sqs.CreateQueueInput{
        QueueName: aws.String("my-queue"),
        Attributes: map[string]*string{
            "Policy": aws.String(fmt.Sprintf(`{
                "Version": "2012-10-17",
                "Statement": [
                    {
                        "Sid": "SNSTopicSendMessage",
                        "Effect": "Allow",
                        "Principal": "*",
                        "Action": "SQS:SendMessage",
                        "Resource": "%s",
                        "Condition": {
                            "ArnEquals": {
                                "aws:SourceArn": "%s"
                            }
                        }
                    }
                ]
            }`, queueARN, b.eventsTopicARN)),
        },
    })
    if err != nil {
        return err
    }
0

Here's an updated example that's a little more fully worked, along with the following changes:

  • Uses the v2 SDK which has context.Context support
  • Sets RawMessageDelivery to true which you probably want when subscripting SQS queues to SNS topics (to preserve message attributes)
  • Fully marshals the policy JSON instead of sprintfing it
  • Doesn't assume any ARNs; in particular, the queue policy doesn't need to know its own ARN (this isn't an IAM role/user policy)
  • Doesn't confirm the subscription, which is only necessary if the topic and queue are in different accounts
import (
    "context"
    "encoding/json"
    "fmt"
    "strings"
    "time"

    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/service/sns"
    "github.com/aws/aws-sdk-go-v2/service/sqs"
    sqstypes "github.com/aws/aws-sdk-go-v2/service/sqs/types"
)

func subscribeSQStoSNS(
    ctx context.Context,
    snsClient *sns.Client, topicARN string,
    sqsClient *sqs.Client, queueName string,
) (*sqs.CreateQueueOutput, *sns.SubscribeOutput, error) {
    createQueueOutput, err := sqsClient.CreateQueue(ctx,
        &sqs.CreateQueueInput{
            QueueName: &queueName,
            Attributes: map[string]string{
                "Policy": queuePolicySNSToSQS(topicARN),
            },
        },
    )
    if err != nil {
        return nil, nil, fmt.Errorf("creating SQS queue %q: %w",
            queueName, err)
    }
    // AWS docs say to wait at least 1 second after creating a queue
    timer := time.NewTimer(1 * time.Second)
    select {
    case <-ctx.Done():
        return nil, nil, fmt.Errorf("waiting after creating SQS queue %q: %w",
            queueName, ctx.Err())
    case <-timer.C:
    }
    queueAttributes, err := sqsClient.GetQueueAttributes(ctx,
        &sqs.GetQueueAttributesInput{
            QueueUrl: createQueueOutput.QueueUrl,
            AttributeNames: []sqstypes.QueueAttributeName{
                sqstypes.QueueAttributeNameQueueArn,
            },
        },
    )
    if err != nil {
        return nil, nil, fmt.Errorf("getting attributes for SQS queue %q: %w",
            queueName, err)
    }
    queueARNKey := string(sqstypes.QueueAttributeNameQueueArn)
    queueARN := queueAttributes.Attributes[queueARNKey]
    if queueARN == "" {
        return nil, nil, fmt.Errorf("SQS queue %q has empty ARN", queueName)
    }
    subscribeOutput, err := snsClient.Subscribe(ctx,
        &sns.SubscribeInput{
            Attributes: map[string]string{
                "RawMessageDelivery": "true",
            },
            Endpoint:              &queueARN,
            Protocol:              aws.String("sqs"),
            ReturnSubscriptionArn: true,
            TopicArn:              &topicARN,
        },
    )
    if err != nil {
        return nil, nil, fmt.Errorf(
            "subscribing SQS queue %q to SNS topic %q: %w",
            queueName, topicARN, err)
    }
    return createQueueOutput, subscribeOutput, nil
}

func queuePolicySNSToSQS(topicARN string) string {
    var buf strings.Builder
    err := json.NewEncoder(&buf).Encode(
        map[string]any{
            "Version": "2012-10-17",
            "Statement": map[string]any{
                "Sid":       "SNSTopicSendMessage",
                "Effect":    "Allow",
                "Principal": "*",
                "Action":    "sqs:SendMessage",
                "Resource":  "*",
                "Condition": map[string]any{
                    "ArnEquals": map[string]any{
                        "aws:SourceArn": topicARN,
                    },
                },
            },
        },
    )
    if err != nil {
        panic(err)
    }
    return buf.String()
}

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.