fix(registry/consul):fix can't get service instance in async mode (#1731)

* fix(registry/consul):fix can't get service instance in async mode

* fix(registry/consul): return error if find service failed
pull/1740/head
letian 3 years ago committed by GitHub
parent 17201cd284
commit 1e749de5db
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 47
      contrib/registry/consul/registry.go

@ -144,33 +144,42 @@ func (r *Registry) Watch(ctx context.Context, name string) (registry.Watcher, er
}
if !ok {
go r.resolve(set)
err := r.resolve(set)
if err != nil {
return nil, err
}
}
return w, nil
}
func (r *Registry) resolve(ss *serviceSet) {
func (r *Registry) resolve(ss *serviceSet) error {
ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
services, idx, err := r.cli.Service(ctx, ss.serviceName, 0, true)
cancel()
if err == nil && len(services) > 0 {
if err != nil {
return err
} else if len(services) > 0 {
ss.broadcast(services)
}
ticker := time.NewTicker(time.Second)
defer ticker.Stop()
for {
<-ticker.C
ctx, cancel := context.WithTimeout(context.Background(), time.Second*120)
tmpService, tmpIdx, err := r.cli.Service(ctx, ss.serviceName, idx, true)
cancel()
if err != nil {
time.Sleep(time.Second)
continue
go func() {
ticker := time.NewTicker(time.Second)
defer ticker.Stop()
for {
<-ticker.C
ctx, cancel := context.WithTimeout(context.Background(), time.Second*120)
tmpService, tmpIdx, err := r.cli.Service(ctx, ss.serviceName, idx, true)
cancel()
if err != nil {
time.Sleep(time.Second)
continue
}
if len(tmpService) != 0 && tmpIdx != idx {
services = tmpService
ss.broadcast(services)
}
idx = tmpIdx
}
if len(tmpService) != 0 && tmpIdx != idx {
services = tmpService
ss.broadcast(services)
}
idx = tmpIdx
}
}()
return nil
}

Loading…
Cancel
Save