gocbcorex/analyticscomponent.go at master · couchbase/gocbcorex · GitHub

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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
package gocbcorex

import (
	"context"
	"net/http"

	"github.com/couchbase/gocbcorex/cbanalyticsx"
	"github.com/couchbase/gocbcorex/cbhttpx"
	"go.uber.org/zap"
)

type AnalyticsQueryOptions = cbanalyticsx.QueryOptions
type AnalyticsQueryResultStream = cbanalyticsx.QueryResultStream

type AnalyticsComponent struct {
	baseHttpComponent

	logger  *zap.Logger
	retries RetryManager
}

type AnalyticsComponentConfig struct {
	HttpRoundTripper http.RoundTripper
	Endpoints        map[string]string
	Authenticator    Authenticator
}

type AnalyticsComponentOptions struct {
	Logger    *zap.Logger
	UserAgent string
}

func OrchestrateAnalyticsEndpoint[RespT any](
	ctx context.Context,
	w *AnalyticsComponent,
	fn func(roundTripper http.RoundTripper, endpoint, username, password string) (RespT, error),
) (RespT, error) {
	roundTripper, _, endpoint, username, password, err := w.SelectEndpoint(nil)
	if err != nil {
		var emptyResp RespT
		return emptyResp, err
	}

	if endpoint == "" {
		var emptyResp RespT
		return emptyResp, serviceNotAvailableError{Service: ServiceTypeAnalytics}
	}

	return fn(roundTripper, endpoint, username, password)
}

func OrchestrateAnalyticsMgmtCall[OptsT any, RespT any](
	ctx context.Context,
	w *AnalyticsComponent,
	execFn func(o cbanalyticsx.Analytics, ctx context.Context, req OptsT) (RespT, error),
	opts OptsT,
) (RespT, error) {
	return OrchestrateRetries(ctx, w.retries, func() (RespT, error) {
		return OrchestrateAnalyticsEndpoint(ctx, w,
			func(roundTripper http.RoundTripper, endpoint, username, password string) (RespT, error) {
				return execFn(cbanalyticsx.Analytics{
					Logger:    w.logger,
					UserAgent: w.userAgent,
					Transport: roundTripper,
					Endpoint:  endpoint,
					Auth: &cbhttpx.BasicAuth{
						Username: username,
						Password: password,
					},
				}, ctx, opts)
			})
	})
}

func OrchestrateNoResAnalyticsMgmtCall[OptsT any](
	ctx context.Context,
	w *AnalyticsComponent,
	execFn func(o cbanalyticsx.Analytics, ctx context.Context, req OptsT) error,
	opts OptsT,
) error {
	return OrchestrateNoResponseRetries(ctx, w.retries, func() error {
		_, err := OrchestrateAnalyticsEndpoint(ctx, w,
			func(roundTripper http.RoundTripper, endpoint, username, password string) (interface{}, error) {
				return nil, execFn(cbanalyticsx.Analytics{
					Logger:    w.logger,
					UserAgent: w.userAgent,
					Transport: roundTripper,
					Endpoint:  endpoint,
					Auth: &cbhttpx.BasicAuth{
						Username: username,
						Password: password,
					},
				}, ctx, opts)
			})
		return err
	})
}

func NewAnalyticsComponent(retries RetryManager, config *AnalyticsComponentConfig, opts *AnalyticsComponentOptions) *AnalyticsComponent {
	return &AnalyticsComponent{
		baseHttpComponent: baseHttpComponent{
			serviceType: ServiceTypeAnalytics,
			userAgent:   opts.UserAgent,
			state: &baseHttpComponentState{
				httpRoundTripper: config.HttpRoundTripper,
				endpoints:        config.Endpoints,
				authenticator:    config.Authenticator,
			},
		},
		logger:  opts.Logger,
		retries: retries,
	}
}

func (w *AnalyticsComponent) Reconfigure(config *AnalyticsComponentConfig) error {
	w.updateState(baseHttpComponentState{
		httpRoundTripper: config.HttpRoundTripper,
		endpoints:        config.Endpoints,
		authenticator:    config.Authenticator,
	})
	return nil
}

type GetAnalyticsEndpointResult struct {
	RoundTripper http.RoundTripper
	Endpoint     string
	Username     string
	Password     string
}

func (w *AnalyticsComponent) GetEndpoint(ctx context.Context) (*GetAnalyticsEndpointResult, error) {
	return OrchestrateAnalyticsEndpoint(ctx, w,
		func(roundTripper http.RoundTripper, endpoint, username, password string) (*GetAnalyticsEndpointResult, error) {
			return &GetAnalyticsEndpointResult{
				RoundTripper: roundTripper,
				Endpoint:     endpoint,
				Username:     username,
				Password:     password,
			}, nil
		})
}

func (w *AnalyticsComponent) Query(ctx context.Context, opts *AnalyticsQueryOptions) (AnalyticsQueryResultStream, error) {
	return OrchestrateRetries(ctx, w.retries, func() (AnalyticsQueryResultStream, error) {
		return OrchestrateAnalyticsEndpoint(ctx, w,
			func(roundTripper http.RoundTripper, endpoint, username, password string) (AnalyticsQueryResultStream, error) {
				return cbanalyticsx.Analytics{
					Logger:    w.logger,
					UserAgent: w.userAgent,
					Transport: roundTripper,
					Endpoint:  endpoint,
					Auth: &cbhttpx.BasicAuth{
						Username: username,
						Password: password,
					},
				}.Query(ctx, opts)
			})
	})
}