/
writer.go
78 lines (69 loc) · 1.72 KB
/
writer.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
/*
Copyright 2021 Adevinta
*/
package sqs
import (
"errors"
"fmt"
"sync"
"github.com/adevinta/vulcan-agent/log"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/arn"
"github.com/aws/aws-sdk-go/aws/session"
"github.com/aws/aws-sdk-go/service/sqs"
"github.com/aws/aws-sdk-go/service/sqs/sqsiface"
)
// Writer writes messages to and AWS SQS queue.
type Writer struct {
*sync.RWMutex
sqs sqsiface.SQSAPI
queueURL string
}
// NewWriter creates a new SQS writer to writer to given queue ARN using the
// passed in endpoint, or the default one if the it is empty.
func NewWriter(queueARN string, endpoint string, log log.Logger) (*Writer, error) {
sess, err := session.NewSession()
if err != nil {
err = fmt.Errorf("creating AWS session %w", err)
return nil, err
}
arn, err := arn.Parse(queueARN)
if err != nil {
err = fmt.Errorf("error parsing SQS queue ARN: %w", err)
return nil, err
}
awsCfg := aws.NewConfig()
if arn.Region != "" {
awsCfg = awsCfg.WithRegion(arn.Region)
}
if endpoint != "" {
awsCfg = awsCfg.WithEndpoint(endpoint)
}
sqsSrv := sqs.New(sess, awsCfg)
params := &sqs.GetQueueUrlInput{
QueueName: aws.String(arn.Resource),
}
if arn.AccountID != "" {
params.SetQueueOwnerAWSAccountId(arn.AccountID)
}
resp, err := sqsSrv.GetQueueUrl(params)
if err != nil {
err = fmt.Errorf("error retrieving SQS queue URL: %w", err)
return nil, err
}
if resp.QueueUrl == nil {
return nil, errors.New("unexpected nill getting queue ARN")
}
return &Writer{
queueURL: *resp.QueueUrl,
sqs: sqsSrv,
}, nil
}
func (w *Writer) Write(body string) error {
msg := &sqs.SendMessageInput{
QueueUrl: &w.queueURL,
MessageBody: &body,
}
_, err := w.sqs.SendMessage(msg)
return err
}