forked from harlow/kinesis-consumer
-
Notifications
You must be signed in to change notification settings - Fork 1
/
emitter.go
43 lines (35 loc) · 1.11 KB
/
emitter.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
package s3
import (
"io"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/session"
"github.com/aws/aws-sdk-go/service/s3"
)
// Emitter stores data in S3 bucket.
//
// The use of this struct requires the configuration of an S3 bucket/endpoint. When the buffer is full, this
// struct's Emit method adds the contents of the buffer to S3 as one file. The filename is generated
// from the first and last sequence numbers of the records contained in that file separated by a
// dash. This struct requires the configuration of an S3 bucket and endpoint.
type Emitter struct {
Bucket string
Region string
}
// Emit is invoked when the buffer is full. This method emits the set of filtered records.
func (e Emitter) Emit(s3Key string, b io.ReadSeeker) error {
svc := s3.New(
session.New(aws.NewConfig().WithMaxRetries(10)),
aws.NewConfig().WithRegion(e.Region),
)
params := &s3.PutObjectInput{
Body: b,
Bucket: aws.String(e.Bucket),
ContentType: aws.String("text/plain"),
Key: aws.String(s3Key),
}
_, err := svc.PutObject(params)
if err != nil {
return err
}
return nil
}