File: resource.go

package info (click to toggle)
golang-github-lestrrat-go-httprc 3.0.4-1
  • links: PTS, VCS
  • area: main
  • in suites: forky
  • size: 316 kB
  • sloc: perl: 60; makefile: 2
file content (366 lines) | stat: -rw-r--r-- 10,895 bytes parent folder | download | duplicates (3)
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
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
package httprc

import (
	"context"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"sync"
	"sync/atomic"
	"time"

	"github.com/lestrrat-go/blackmagic"
	"github.com/lestrrat-go/httpcc"
	"github.com/lestrrat-go/httprc/v3/tracesink"
)

// ResourceBase is a generic Resource type
type ResourceBase[T any] struct {
	u           string
	ready       chan struct{} // closed when the resource is ready (i.e. after first successful fetch)
	once        sync.Once
	httpcl      HTTPClient
	t           Transformer[T]
	r           atomic.Value
	next        atomic.Value
	interval    time.Duration
	minInterval atomic.Int64
	maxInterval atomic.Int64
	busy        atomic.Bool
}

// NewResource creates a new Resource object which after fetching the
// resource from the URL, will transform the response body using the
// provided Transformer to an object of type T.
//
// This function will return an error if the URL is not a valid URL
// (i.e. it cannot be parsed by url.Parse), or if the transformer is nil.
func NewResource[T any](s string, transformer Transformer[T], options ...NewResourceOption) (*ResourceBase[T], error) {
	var httpcl HTTPClient
	var interval time.Duration
	minInterval := DefaultMinInterval
	maxInterval := DefaultMaxInterval
	for _, option := range options {
		switch option.Ident() {
		case identHTTPClient{}:
			if err := option.Value(&httpcl); err != nil {
				return nil, fmt.Errorf(`httprc.NewResource: failed to parse HTTPClient option: %w`, err)
			}
		case identMinimumInterval{}:
			if err := option.Value(&minInterval); err != nil {
				return nil, fmt.Errorf(`httprc.NewResource: failed to parse MinimumInterval option: %w`, err)
			}
		case identMaximumInterval{}:
			if err := option.Value(&maxInterval); err != nil {
				return nil, fmt.Errorf(`httprc.NewResource: failed to parse MaximumInterval option: %w`, err)
			}
		case identConstantInterval{}:
			if err := option.Value(&interval); err != nil {
				return nil, fmt.Errorf(`httprc.NewResource: failed to parse ConstantInterval option: %w`, err)
			}
		}
	}
	if transformer == nil {
		return nil, fmt.Errorf(`httprc.NewResource: %w`, errTransformerRequired)
	}

	if s == "" {
		return nil, fmt.Errorf(`httprc.NewResource: %w`, errURLCannotBeEmpty)
	}

	if _, err := url.Parse(s); err != nil {
		return nil, fmt.Errorf(`httprc.NewResource: %w`, err)
	}
	r := &ResourceBase[T]{
		u:        s,
		httpcl:   httpcl,
		t:        transformer,
		interval: interval,
		ready:    make(chan struct{}),
	}
	if httpcl != nil {
		r.httpcl = httpcl
	}
	r.minInterval.Store(int64(minInterval))
	r.maxInterval.Store(int64(maxInterval))
	r.SetNext(time.Unix(0, 0)) // initially, it should be fetched immediately
	return r, nil
}

// URL returns the URL of the resource.
func (r *ResourceBase[T]) URL() string {
	return r.u
}

// Ready returns an empty error when the resource is ready. If the context
// is canceled before the resource is ready, it will return the error from
// the context.
func (r *ResourceBase[T]) Ready(ctx context.Context) error {
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-r.ready:
		return nil
	}
}

// Get assigns the value of the resource to the provided pointer.
// If using the `httprc.ResourceBase[T]` type directly, you can use the `Resource()`
// method to get the resource directly.
//
// This method exists because parametric types cannot be assigned to a single object type
// that return different return values of the specialized type. i.e. for resources
// `ResourceBase[A]` and `ResourceBase[B]`, we cannot have a single interface that can
// be assigned to the same interface type `X` that expects a `Resource()` method that
// returns `A` or `B` depending on the type of the resource. When accessing the
// resource through the `httprc.Resource` interface, use this method to obtain the
// stored value.
func (r *ResourceBase[T]) Get(dst any) error {
	return blackmagic.AssignIfCompatible(dst, r.Resource())
}

// Resource returns the last fetched resource. If the resource has not been
// fetched yet, this will return the zero value of type T.
//
// If you would rather wait until the resource is fetched, you can use the
// `Ready()` method to wait until the resource is ready (i.e. fetched at least once).
func (r *ResourceBase[T]) Resource() T {
	v := r.r.Load()
	switch v := v.(type) {
	case T:
		return v
	default:
		var zero T
		return zero
	}
}

func (r *ResourceBase[T]) Next() time.Time {
	//nolint:forcetypeassert
	return r.next.Load().(time.Time)
}

func (r *ResourceBase[T]) SetNext(v time.Time) {
	r.next.Store(v)
}

func (r *ResourceBase[T]) ConstantInterval() time.Duration {
	return r.interval
}

func (r *ResourceBase[T]) MaxInterval() time.Duration {
	return time.Duration(r.maxInterval.Load())
}

func (r *ResourceBase[T]) MinInterval() time.Duration {
	return time.Duration(r.minInterval.Load())
}

func (r *ResourceBase[T]) SetMaxInterval(v time.Duration) {
	r.maxInterval.Store(int64(v))
}

func (r *ResourceBase[T]) SetMinInterval(v time.Duration) {
	r.minInterval.Store(int64(v))
}

func (r *ResourceBase[T]) SetBusy(v bool) {
	r.busy.Store(v)
}

func (r *ResourceBase[T]) IsBusy() bool {
	return r.busy.Load()
}

// limitedBody is a wrapper around an io.Reader that will only read up to
// MaxBufferSize bytes. This is provided to prevent the user from accidentally
// reading a huge response body into memory
type limitedBody struct {
	rdr   io.Reader
	close func() error
}

func (l *limitedBody) Read(p []byte) (n int, err error) {
	return l.rdr.Read(p)
}

func (l *limitedBody) Close() error {
	return l.close()
}

type traceSinkKey struct{}

func withTraceSink(ctx context.Context, sink TraceSink) context.Context {
	return context.WithValue(ctx, traceSinkKey{}, sink)
}

func traceSinkFromContext(ctx context.Context) TraceSink {
	if v := ctx.Value(traceSinkKey{}); v != nil {
		//nolint:forcetypeassert
		return v.(TraceSink)
	}
	return tracesink.Nop{}
}

type httpClientKey struct{}

func withHTTPClient(ctx context.Context, cl HTTPClient) context.Context {
	return context.WithValue(ctx, httpClientKey{}, cl)
}

func httpClientFromContext(ctx context.Context) HTTPClient {
	if v := ctx.Value(httpClientKey{}); v != nil {
		//nolint:forcetypeassert
		return v.(HTTPClient)
	}
	return http.DefaultClient
}

func (r *ResourceBase[T]) Sync(ctx context.Context) error {
	traceSink := traceSinkFromContext(ctx)
	httpcl := r.httpcl
	if httpcl == nil {
		httpcl = httpClientFromContext(ctx)
	}

	req, err := http.NewRequestWithContext(ctx, http.MethodGet, r.u, nil)
	if err != nil {
		return fmt.Errorf(`httprc.Resource.Sync: failed to create request: %w`, err)
	}

	traceSink.Put(ctx, fmt.Sprintf("httprc.Resource.Sync: fetching %q", r.u))
	res, err := httpcl.Do(req)
	if err != nil {
		return fmt.Errorf(`httprc.Resource.Sync: failed to execute HTTP request: %w`, err)
	}
	defer res.Body.Close()

	next := r.calculateNextRefreshTime(ctx, res)
	traceSink.Put(ctx, fmt.Sprintf("httprc.Resource.Sync: next refresh time for %q is %v", r.u, next))
	r.SetNext(next)

	if res.StatusCode != http.StatusOK {
		return fmt.Errorf(`httprc.Resource.Sync: %w (status code=%d, url=%q)`, errUnexpectedStatusCode, res.StatusCode, r.u)
	}

	// replace the body of the response with a limited reader that
	// will only read up to MaxBufferSize bytes
	res.Body = &limitedBody{
		rdr:   &io.LimitedReader{R: res.Body, N: MaxBufferSize},
		close: res.Body.Close,
	}
	traceSink.Put(ctx, fmt.Sprintf("httprc.Resource.Sync: transforming %q", r.u))
	v, err := r.transform(ctx, res)
	if err != nil {
		return fmt.Errorf(`httprc.Resource.Sync: %w: %w`, errTransformerFailed, err)
	}

	traceSink.Put(ctx, fmt.Sprintf("httprc.Resource.Sync: storing new value for %q", r.u))
	r.r.Store(v)
	r.once.Do(func() { close(r.ready) })
	traceSink.Put(ctx, fmt.Sprintf("httprc.Resource.Sync: stored value for %q", r.u))
	return nil
}

func (r *ResourceBase[T]) transform(ctx context.Context, res *http.Response) (ret T, gerr error) {
	// Protect the call to Transform with a defer/recover block, so that even
	// if the Transform method panics, we can recover from it and return an error
	defer func() {
		if recovered := recover(); recovered != nil {
			gerr = fmt.Errorf(`httprc.Resource.transform: %w: %v`, errRecoveredFromPanic, recovered)
		}
	}()
	return r.t.Transform(ctx, res)
}

func (r *ResourceBase[T]) determineNextFetchInterval(ctx context.Context, name string, fromHeader, minValue, maxValue time.Duration) time.Duration {
	traceSink := traceSinkFromContext(ctx)

	if fromHeader > maxValue {
		traceSink.Put(ctx, fmt.Sprintf("httprc.Resource.Sync: %s %s > maximum interval, using maximum interval %s", r.URL(), name, maxValue))
		return maxValue
	}

	if fromHeader < minValue {
		traceSink.Put(ctx, fmt.Sprintf("httprc.Resource.Sync: %s %s < minimum interval, using minimum interval %s", r.URL(), name, minValue))
		return minValue
	}

	traceSink.Put(ctx, fmt.Sprintf("httprc.Resource.Sync: %s Using %s (%s)", r.URL(), name, fromHeader))
	return fromHeader
}

func (r *ResourceBase[T]) calculateNextRefreshTime(ctx context.Context, res *http.Response) time.Time {
	traceSink := traceSinkFromContext(ctx)
	now := time.Now()

	// If constant interval is set, use that regardless of what the
	// response headers say.
	if interval := r.ConstantInterval(); interval > 0 {
		traceSink.Put(ctx, fmt.Sprintf("httprc.Resource.Sync: %s Explicit interval set, using value %s", r.URL(), interval))
		return now.Add(interval)
	}

	if interval := r.extractCacheControlMaxAge(ctx, res); interval > 0 {
		return now.Add(interval)
	}

	if interval := r.extractExpiresInterval(ctx, res); interval > 0 {
		return now.Add(interval)
	}

	traceSink.Put(ctx, fmt.Sprintf("httprc.Resource.Sync: %s No cache-control/expires headers found, using minimum interval", r.URL()))
	return now.Add(r.MinInterval())
}

func (r *ResourceBase[T]) extractCacheControlMaxAge(ctx context.Context, res *http.Response) time.Duration {
	traceSink := traceSinkFromContext(ctx)

	v := res.Header.Get(`Cache-Control`)
	if v == "" {
		return 0
	}

	dir, err := httpcc.ParseResponse(v)
	if err != nil {
		return 0
	}

	maxAge, ok := dir.MaxAge()
	if !ok {
		return 0
	}

	traceSink.Put(ctx, fmt.Sprintf("httprc.Resource.Sync: %s Cache-Control=max-age directive set (%d)", r.URL(), maxAge))
	return r.determineNextFetchInterval(
		ctx,
		"max-age",
		time.Duration(maxAge)*time.Second,
		r.MinInterval(),
		r.MaxInterval(),
	)
}

func (r *ResourceBase[T]) extractExpiresInterval(ctx context.Context, res *http.Response) time.Duration {
	traceSink := traceSinkFromContext(ctx)

	v := res.Header.Get(`Expires`)
	if v == "" {
		return 0
	}

	expires, err := http.ParseTime(v)
	if err != nil {
		return 0
	}

	traceSink.Put(ctx, fmt.Sprintf("httprc.Resource.Sync: %s Expires header set (%s)", r.URL(), expires))
	return r.determineNextFetchInterval(
		ctx,
		"expires",
		time.Until(expires),
		r.MinInterval(),
		r.MaxInterval(),
	)
}