-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathfactory.go
More file actions
119 lines (107 loc) · 3.08 KB
/
Copy pathfactory.go
File metadata and controls
119 lines (107 loc) · 3.08 KB
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
package bqin
import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/credentials"
"github.com/aws/aws-sdk-go/aws/endpoints"
"github.com/aws/aws-sdk-go/aws/session"
"google.golang.org/api/option"
)
type Factory struct {
*Config
}
func (f *Factory) NewAWSSession() *session.Session {
c := f.Config.Cloud.AWS
conf := &aws.Config{
DisableSSL: aws.Bool(c.DisableSSL),
Region: aws.String(c.Region),
S3ForcePathStyle: aws.Bool(c.S3ForcePathStyle),
}
if c.S3Endpoint != "" || c.SQSEndpoint != "" {
defaultResolver := endpoints.DefaultResolver()
customResolver := endpoints.ResolverFunc(
func(service, region string, optFns ...func(*endpoints.Options)) (endpoints.ResolvedEndpoint, error) {
if c.S3Endpoint != "" && service == endpoints.S3ServiceID {
return endpoints.ResolvedEndpoint{
URL: c.S3Endpoint,
SigningRegion: region,
}, nil
}
if c.SQSEndpoint != "" && service == endpoints.SqsServiceID {
return endpoints.ResolvedEndpoint{
URL: c.SQSEndpoint,
SigningRegion: region,
}, nil
}
return defaultResolver.EndpointFor(service, region, optFns...)
},
)
conf = conf.WithEndpointResolver(customResolver)
}
if c.AccessKeyID != "" || c.SecretAccessKey != "" {
conf = conf.WithCredentials(credentials.NewStaticCredentials(c.AccessKeyID, c.SecretAccessKey, ""))
}
var shardConfigState session.SharedConfigState = session.SharedConfigEnable
if c.DisableShardConfigState {
shardConfigState = session.SharedConfigDisable
}
sess := session.Must(session.NewSessionWithOptions(session.Options{
Config: *conf,
SharedConfigState: shardConfigState,
}))
return sess
}
func (f *Factory) NewCloudStorageOptions() []option.ClientOption {
opts := f.NewGCPOptions()
if endpoint := f.Config.Cloud.GCP.CloudStorageEndpoint; endpoint != "" {
opts = append(opts, option.WithEndpoint(endpoint))
}
return opts
}
func (f *Factory) NewBigQueryOptions() []option.ClientOption {
opts := f.NewGCPOptions()
if endpoint := f.Config.Cloud.GCP.BigQueryEndpoint; endpoint != "" {
opts = append(opts, option.WithEndpoint(endpoint))
}
return opts
}
func (f *Factory) NewGCPOptions() []option.ClientOption {
c := f.Config.Cloud.GCP
opts := make([]option.ClientOption, 0, 2)
if c.WithoutAuthentication {
opts = append(opts, option.WithoutAuthentication())
}
if !c.Base64Credential.IsEmpty() {
opts = append(opts, option.WithCredentialsJSON(c.Base64Credential.Bytes()))
}
return opts
}
func (f *Factory) NewReceiver() *Receiver {
return NewReceiver(
f.Config.QueueName,
f.NewAWSSession(),
)
}
func (f *Factory) NewResolver() *Resolver {
return NewResolver(
f.Config.Rules,
)
}
func (f *Factory) NewTransporter() *Transporter {
return NewTransporter(
f.NewAWSSession(),
f.NewCloudStorageOptions()...,
)
}
func (f *Factory) NewLoader() *Loader {
return NewLoader(
f.NewBigQueryOptions()...,
)
}
func (f *Factory) NewApp() *App {
return &App{
Receiver: f.NewReceiver(),
Resolver: f.NewResolver(),
Transporter: f.NewTransporter(),
Loader: f.NewLoader(),
}
}