forked from go-kit/kit
/
subscriber.go
100 lines (88 loc) · 2.22 KB
/
subscriber.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
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
package dnssrv
import (
"fmt"
"net"
"time"
"github.com/go-kit/kit/endpoint"
"github.com/go-kit/kit/log"
"github.com/go-kit/kit/sd"
"github.com/go-kit/kit/sd/cache"
)
// Subscriber yields endpoints taken from the named DNS SRV record. The name is
// resolved on a fixed schedule. Priorities and weights are ignored.
type Subscriber struct {
name string
cache *cache.Cache
logger log.Logger
quit chan struct{}
}
// NewSubscriber returns a DNS SRV subscriber.
func NewSubscriber(
name string,
ttl time.Duration,
factory sd.Factory,
logger log.Logger,
) *Subscriber {
return NewSubscriberDetailed(name, time.NewTicker(ttl), net.LookupSRV, factory, logger)
}
// NewSubscriberDetailed is the same as NewSubscriber, but allows users to
// provide an explicit lookup refresh ticker instead of a TTL, and specify the
// lookup function instead of using net.LookupSRV.
func NewSubscriberDetailed(
name string,
refresh *time.Ticker,
lookup Lookup,
factory sd.Factory,
logger log.Logger,
) *Subscriber {
p := &Subscriber{
name: name,
cache: cache.New(factory, logger),
logger: logger,
quit: make(chan struct{}),
}
instances, err := p.resolve(lookup)
if err == nil {
logger.Log("name", name, "instances", len(instances))
} else {
logger.Log("name", name, "err", err)
}
p.cache.Update(instances)
go p.loop(refresh, lookup)
return p
}
// Stop terminates the Subscriber.
func (p *Subscriber) Stop() {
close(p.quit)
}
func (p *Subscriber) loop(t *time.Ticker, lookup Lookup) {
defer t.Stop()
for {
select {
case <-t.C:
instances, err := p.resolve(lookup)
if err != nil {
p.logger.Log("name", p.name, "err", err)
continue // don't replace potentially-good with bad
}
p.cache.Update(instances)
case <-p.quit:
return
}
}
}
// Endpoints implements the Subscriber interface.
func (p *Subscriber) Endpoints() ([]endpoint.Endpoint, error) {
return p.cache.Endpoints(), nil
}
func (p *Subscriber) resolve(lookup Lookup) ([]string, error) {
_, addrs, err := lookup("", "", p.name)
if err != nil {
return []string{}, err
}
instances := make([]string, len(addrs))
for i, addr := range addrs {
instances[i] = net.JoinHostPort(addr.Target, fmt.Sprint(addr.Port))
}
return instances, nil
}