-
Notifications
You must be signed in to change notification settings - Fork 0
/
deliverystream.go
76 lines (59 loc) · 1.88 KB
/
deliverystream.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
package firehose
import (
"context"
"fmt"
"github.com/observeinc/aws-snapshot/pkg/api"
"github.com/aws/aws-sdk-go/service/firehose"
)
type DescribeDeliveryStreamOutput struct {
*firehose.DescribeDeliveryStreamOutput
}
func (o *DescribeDeliveryStreamOutput) Records() (records []*api.Record) {
if desc := o.DeliveryStreamDescription; desc != nil {
records = append(records, &api.Record{
ID: o.DeliveryStreamDescription.DeliveryStreamARN,
Data: o.DeliveryStreamDescription,
})
}
return
}
type DescribeDeliveryStreams struct {
API
}
var _ api.RequestBuilder = &DescribeDeliveryStreams{}
// New implements api.RequestBuilder
func (fn *DescribeDeliveryStreams) New(name string, config interface{}) ([]api.Request, error) {
var input firehose.ListDeliveryStreamsInput
if err := api.DecodeConfig(config, &input); err != nil {
return nil, err
}
call := func(ctx context.Context, ch chan<- *api.Record) error {
var lastPage bool
for !lastPage {
listOutput, err := fn.ListDeliveryStreamsWithContext(ctx, &input)
if err != nil {
return fmt.Errorf("failed to list streams: %w", err)
}
if len(listOutput.DeliveryStreamNames) == 0 {
break
}
for _, deliveryStreamName := range listOutput.DeliveryStreamNames {
describeDeliveryStreamOutput, err := fn.DescribeDeliveryStreamWithContext(ctx, &firehose.DescribeDeliveryStreamInput{
DeliveryStreamName: deliveryStreamName,
})
if err != nil {
return fmt.Errorf("failed to describe stream %q: %w", *deliveryStreamName, err)
}
if err := api.SendRecords(ctx, ch, name, &DescribeDeliveryStreamOutput{describeDeliveryStreamOutput}); err != nil {
return err
}
input.SetExclusiveStartDeliveryStreamName(*deliveryStreamName)
}
if listOutput.HasMoreDeliveryStreams != nil {
lastPage = !(*listOutput.HasMoreDeliveryStreams)
}
}
return nil
}
return []api.Request{call}, nil
}