Mirroristas/etcd
2
0
Fork 0
mirror of https://github.com/etcd-io/etcd.git synced 2024-09-27 06:25:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #8728 from gyuho/eee

Browse Source 
clientv3: remove balancer interface
This commit is contained in:
Gyu-Ho Lee 2017-10-20 16:43:32 -07:00 committed by GitHub
commit 785a5a11ed
3 changed files with 14 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
clientv3/balancer.go
View File

@ -36,28 +36,6 @@ const (
notifyNext notifyNext
) )
type balancer interface {
grpc.Balancer
ConnectNotify() <-chan struct{}
endpoint(hostPort string) string
endpoints() []string
// pinned returns the current pinned endpoint.
pinned() string
// hostPortError handles error from server-side.
hostPortError(hostPort string, err error)
// up is Up but includes whether the balancer will use the connection.
up(addr grpc.Address) (func(error), bool)
// updateAddrs changes the balancer's endpoints.
updateAddrs(endpoints ...string)
// ready returns a channel that closes when the balancer first connects.
ready() <-chan struct{}
// next forces the balancer to switch endpoints.
next()
}
// simpleBalancer does the bare minimum to expose multiple eps // simpleBalancer does the bare minimum to expose multiple eps
// to the grpc reconnection code path // to the grpc reconnection code path
type simpleBalancer struct { type simpleBalancer struct {
@ -152,8 +130,6 @@ func (b *simpleBalancer) pinned() string {
return b.pinAddr return b.pinAddr
} }
func (b *simpleBalancer) hostPortError(hostPort string, err error) { return }
func getHostPort2ep(eps []string) map[string]string { func getHostPort2ep(eps []string) map[string]string {
hm := make(map[string]string, len(eps)) hm := make(map[string]string, len(eps))
for i := range eps { for i := range eps {

2
clientv3/client.go
View File

@ -55,7 +55,7 @@ type Client struct {
cfg Config cfg Config
creds *credentials.TransportCredentials creds *credentials.TransportCredentials
balancer balancer balancer *healthBalancer
mu sync.Mutex mu sync.Mutex
ctx context.Context ctx context.Context

14
clientv3/health_balancer.go
View File

@ -33,7 +33,7 @@ type healthCheckFunc func(ep string) (bool, error)
// healthBalancer wraps a balancer so that it uses health checking // healthBalancer wraps a balancer so that it uses health checking
// to choose its endpoints. // to choose its endpoints.
type healthBalancer struct { type healthBalancer struct {
balancer *simpleBalancer
// healthCheck checks an endpoint's health. // healthCheck checks an endpoint's health.
healthCheck healthCheckFunc healthCheck healthCheckFunc
@ -59,9 +59,9 @@ type healthBalancer struct {
wg sync.WaitGroup wg sync.WaitGroup
} }
func newHealthBalancer(b balancer, timeout time.Duration, hc healthCheckFunc) *healthBalancer { func newHealthBalancer(b *simpleBalancer, timeout time.Duration, hc healthCheckFunc) *healthBalancer {
hb := &healthBalancer{ hb := &healthBalancer{
balancer: b, simpleBalancer: b,
healthCheck: hc, healthCheck: hc,
eps: b.endpoints(), eps: b.endpoints(),
addrs: eps2addrs(b.endpoints()), addrs: eps2addrs(b.endpoints()),
@ -107,13 +107,13 @@ func (hb *healthBalancer) up(addr grpc.Address) (func(error), bool) {
if !hb.mayPin(addr) { if !hb.mayPin(addr) {
return func(err error) {}, false return func(err error) {}, false
} }
return hb.balancer.up(addr) return hb.simpleBalancer.up(addr)
} }
func (hb *healthBalancer) Close() error { func (hb *healthBalancer) Close() error {
hb.stopOnce.Do(func() { close(hb.stopc) }) hb.stopOnce.Do(func() { close(hb.stopc) })
hb.wg.Wait() hb.wg.Wait()
return hb.balancer.Close() return hb.simpleBalancer.Close()
} }
func (hb *healthBalancer) updateAddrs(eps ...string) { func (hb *healthBalancer) updateAddrs(eps ...string) {
@ -122,7 +122,7 @@ func (hb *healthBalancer) updateAddrs(eps ...string) {
hb.addrs, hb.eps, hb.hostPort2ep = addrs, eps, hostPort2ep hb.addrs, hb.eps, hb.hostPort2ep = addrs, eps, hostPort2ep
hb.unhealthy = make(map[string]time.Time) hb.unhealthy = make(map[string]time.Time)
hb.mu.Unlock() hb.mu.Unlock()
hb.balancer.updateAddrs(eps...) hb.simpleBalancer.updateAddrs(eps...)
} }
func (hb *healthBalancer) endpoint(host string) string { func (hb *healthBalancer) endpoint(host string) string {
@ -162,7 +162,7 @@ func (hb *healthBalancer) updateUnhealthy(timeout time.Duration) {
for _, addr := range hb.liveAddrs() { for _, addr := range hb.liveAddrs() {
eps = append(eps, hb.endpoint(addr.Addr)) eps = append(eps, hb.endpoint(addr.Addr))
} }
hb.balancer.updateAddrs(eps...) hb.simpleBalancer.updateAddrs(eps...)
case <-hb.stopc: case <-hb.stopc:
return return
} }