-
Notifications
You must be signed in to change notification settings - Fork 2
/
client.go
79 lines (60 loc) · 2.1 KB
/
client.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
79
package sqs
import (
"context"
"github.com/americanas-go/errors"
"github.com/americanas-go/log"
"github.com/aws/aws-sdk-go-v2/service/sqs"
"github.com/aws/aws-sdk-go/aws"
)
// Client knows how to publish on sqs.
type Client interface {
// Publish publishes message on sns.
Publish(ctx context.Context, input *sqs.SendMessageInput) error
// ResolveQueueUrl resolves the URL of the queue.
ResolveQueueUrl(ctx context.Context, queueName string) (*string, error)
}
type sqsClient interface {
SendMessage(ctx context.Context, params *sqs.SendMessageInput, optFns ...func(*sqs.Options)) (*sqs.SendMessageOutput, error)
GetQueueUrl(ctx context.Context, params *sqs.GetQueueUrlInput, optFns ...func(*sqs.Options)) (*sqs.GetQueueUrlOutput, error)
}
// Client holds client and resource name.
type client struct {
client sqsClient
queueUrls map[string]*string
}
// NewClient returns a initialized client.
func NewClient(c *sqs.Client) Client {
return &client{c, map[string]*string{}}
}
// Publish publishes input message to the configured sqs queue.
func (c *client) Publish(ctx context.Context, input *sqs.SendMessageInput) error {
logger := log.FromContext(ctx).
WithTypeOf(*c).
WithField("subject", input.QueueUrl)
logger.Tracef("sending message to sqs")
response, err := c.client.SendMessage(ctx, input)
if err != nil {
return errors.Wrap(err, errors.New("error sending message to sqs"))
}
logger.
WithField("message_id", *response.MessageId).
Debug("message sent to sqs")
return nil
}
// ResolveQueueUrl resolves sqs queue url according to queueName.
func (c *client) ResolveQueueUrl(ctx context.Context, queueName string) (*string, error) {
if queueUrl, ok := c.queueUrls[queueName]; ok {
return queueUrl, nil
}
result, err := c.client.GetQueueUrl(ctx, &sqs.GetQueueUrlInput{
QueueName: aws.String(queueName),
})
if err != nil {
return nil, errors.Wrap(err, errors.New("error resolving sqs queue url"))
}
if result == nil || result.QueueUrl == nil {
return nil, errors.Errorf("sqs queue %s not found", queueName)
}
c.queueUrls[queueName] = result.QueueUrl
return result.QueueUrl, nil
}