-
Notifications
You must be signed in to change notification settings - Fork 0
/
sinkers.go
129 lines (109 loc) · 3.36 KB
/
sinkers.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
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
package sqsdr
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"log"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/sqs"
"github.com/aws/aws-sdk-go/service/sqs/sqsiface"
)
// Sinker is an interface that accepts an array of SQS messages and puts them
// somewhere. Where "somewhere" could be into another SQS queue, a file on disk,
// or whatever.
type Sinker interface {
Sink(context.Context, []*sqs.Message) error
}
// NoOpSink drops the messages on the floor. Use it only as a signal to other developers
// that your other sink is doing all of the work.
type NoOpSink struct{}
// Sink does nothing with the messages and returns a nil error
func (n NoOpSink) Sink(ctx context.Context, msgs []*sqs.Message) error {
return nil
}
// SQSSink pass all messages to the QueueURL with the provided SQS Client
type SQSSink struct {
QueueURL string
Client sqsiface.SQSAPI
}
// Sink performs a BatchSend with the passed in messages
func (s *SQSSink) Sink(ctx context.Context, msgs []*sqs.Message) error {
entries := make([]*sqs.SendMessageBatchRequestEntry, len(msgs))
for i, msg := range msgs {
entry := &sqs.SendMessageBatchRequestEntry{
Id: msg.MessageId,
MessageAttributes: msg.MessageAttributes,
MessageBody: msg.Body,
}
entries[i] = entry
}
resp, err := s.Client.SendMessageBatchWithContext(
ctx,
&sqs.SendMessageBatchInput{
QueueUrl: aws.String(s.QueueURL),
Entries: entries,
},
)
if err != nil {
return err
}
if len(resp.Failed) > 0 {
var errBuffer bytes.Buffer
errBuffer.WriteString("The following error messages were received:\n\n")
for _, f := range resp.Failed {
errBuffer.WriteString(
fmt.Sprintf("%v\n\n", *f.Message),
)
}
return fmt.Errorf("failed to batch send messages: %v", errBuffer.String())
}
return nil
}
// MessageOutput is a simplified version of the SQS Message that's appropriate to write to disk or
// STDOUT.
//
// TODO: will have to handle the case when putting messages from STDIN to queue
// where the message body is JSON.
type MessageOutput struct {
Body *string `json:",omitempty"`
MessageAttributes map[string]*sqs.MessageAttributeValue `json:",omitempty"`
MessageId *string
ReceiptHandle *string
}
// WriterSink will write SQS Message in the MessageOutput format to the Writer with the delimiter
// as a separator
type WriterSink struct {
Writer io.Writer
Passthrough Sinker
}
// Sink writes converts the SQS Message to a MessageOutput and writes the message
// the Writer.
func (w *WriterSink) Sink(ctx context.Context, msgs []*sqs.Message) error {
errors := make([]error, 0)
encoder := json.NewEncoder(w.Writer)
for _, msg := range msgs {
msgOut := MessageOutput{
Body: msg.Body,
MessageAttributes: msg.MessageAttributes,
MessageId: msg.MessageId,
ReceiptHandle: msg.ReceiptHandle,
}
err := encoder.Encode(msgOut)
if err != nil {
log.Println("an error occurred while dumping SQS message to JSON:", err)
errors = append(errors, err)
continue
}
}
if len(errors) > 0 {
var buffer bytes.Buffer
for _, e := range errors {
buffer.WriteString(e.Error())
buffer.WriteString("\n===\n")
}
return fmt.Errorf("the following errors occurred while dumping SQS messages: %v", buffer.String())
}
return w.Passthrough.Sink(ctx, msgs)
}