fix http resovler bug (#1231)

* fix http resovler bug

* delete unused code
pull/1235/head
longxboy 3 years ago committed by GitHub
parent 38c9def445
commit dca963a236
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 100
      transport/http/resolver.go

@ -4,6 +4,7 @@ import (
"context" "context"
"net/url" "net/url"
"sync" "sync"
"time"
"github.com/go-kratos/kratos/v2/log" "github.com/go-kratos/kratos/v2/log"
"github.com/go-kratos/kratos/v2/registry" "github.com/go-kratos/kratos/v2/registry"
@ -36,8 +37,9 @@ func parseTarget(endpoint string) (*Target, error) {
} }
type resolver struct { type resolver struct {
lock sync.RWMutex lock sync.RWMutex
nodes []*registry.ServiceInstance nodes []*registry.ServiceInstance
updater Updater
target *Target target *Target
watcher registry.Watcher watcher registry.Watcher
@ -53,69 +55,73 @@ func newResolver(ctx context.Context, discovery registry.Discovery, target *Targ
target: target, target: target,
watcher: watcher, watcher: watcher,
logger: log.NewHelper(log.DefaultLogger), logger: log.NewHelper(log.DefaultLogger),
updater: updater,
} }
done := make(chan error, 1) if block {
go func() { done := make(chan error, 1)
for { go func() {
var executed bool for {
services, err := watcher.Next() services, err := watcher.Next()
if err != nil {
r.logger.Errorf("http client watch service %v got unexpected error:=%v", target, err)
if block {
select {
case done <- err:
default:
}
}
return
}
var nodes []*registry.ServiceInstance
for _, in := range services {
_, endpoint, err := parseEndpoint(in.Endpoints)
if err != nil { if err != nil {
r.logger.Errorf("Failed to parse (%v) discovery endpoint: %v error %v", target, in.Endpoints, err) done <- err
continue return
} }
if endpoint == "" { r.update(services)
continue if len(r.nodes) > 0 {
}
nodes = append(nodes, in)
}
if len(nodes) != 0 {
updater.Update(nodes)
r.lock.Lock()
r.nodes = nodes
r.lock.Unlock()
if block && !executed {
executed = true
done <- nil done <- nil
return
} }
} else {
r.logger.Warnf("[http resovler]Zero endpoint found,refused to write,ser: %s ins: %v", target.Endpoint, nodes)
} }
} }()
}()
if block {
select { select {
case e := <-done: case err := <-done:
if e != nil { if err != nil {
watcher.Stop() watcher.Stop()
return nil, err
} }
return r, e
case <-ctx.Done(): case <-ctx.Done():
r.logger.Errorf("http client watch service %v reaching context deadline!", target) r.logger.Errorf("http client watch service %v reaching context deadline!", target)
watcher.Stop() watcher.Stop()
return nil, ctx.Err() return nil, ctx.Err()
} }
} }
go func() {
for {
services, err := watcher.Next()
if err != nil {
r.logger.Errorf("http client watch service %v got unexpected error:=%v", target, err)
time.Sleep(time.Second)
continue
}
r.update(services)
}
}()
return r, nil return r, nil
} }
func (r *resolver) fetch(ctx context.Context) []*registry.ServiceInstance { func (r *resolver) update(services []*registry.ServiceInstance) {
r.lock.RLock() var nodes []*registry.ServiceInstance
nodes := r.nodes for _, in := range services {
r.lock.RUnlock() _, endpoint, err := parseEndpoint(in.Endpoints)
return nodes if err != nil {
r.logger.Errorf("Failed to parse (%v) discovery endpoint: %v error %v", r.target, in.Endpoints, err)
continue
}
if endpoint == "" {
continue
}
nodes = append(nodes, in)
}
if len(nodes) != 0 {
r.updater.Update(nodes)
r.lock.Lock()
r.nodes = nodes
r.lock.Unlock()
} else {
r.logger.Warnf("[http resovler]Zero endpoint found,refused to write,ser: %s ins: %v", r.target.Endpoint, nodes)
}
} }
func parseEndpoint(endpoints []string) (string, string, error) { func parseEndpoint(endpoints []string) (string, string, error) {

Loading…
Cancel
Save