-
Notifications
You must be signed in to change notification settings - Fork 0
/
dump.go
46 lines (38 loc) · 1.03 KB
/
dump.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
package sqsdr
import (
"io"
"github.com/aws/aws-sdk-go/service/sqs/sqsiface"
)
// Dump persists all messages in a queue to disk and then makes sure all of the
// messages end up back in the source queue.
type Dump struct {
SourceClient sqsiface.SQSAPI
SourceQueueURL string
Out io.Writer
}
// Dump uses a FallthroughPipeline to place all messages in a temporary queue after
// they've been written to disk. The messages will be placed back into the source
// queue using a pipline in the reverse direction.
func (d *Dump) Dump() error {
chooser := &RightPassthroughChooser{}
leftSink := &NoOpSink{}
rightSinkFunc := func(queueURL string, client sqsiface.SQSAPI) Sinker {
pass := &SQSSink{
QueueURL: queueURL,
Client: client,
}
w := &WriterSink{
Writer: d.Out,
Passthrough: pass,
}
return w
}
f := &FallthroughPipeline{
Chooser: chooser,
LeftSink: leftSink,
RightSinkFunc: rightSinkFunc,
SourceClient: d.SourceClient,
SourceQueueURL: d.SourceQueueURL,
}
return f.Run()
}