代码拉取完成,页面将自动刷新
package discovery
import (
"context"
"encoding/json"
"errors"
"gitee.com/llakcs/agile-go/log"
"gitee.com/llakcs/agile-go/registry"
"gitee.com/llakcs/agile-go/utils/endpoint"
"time"
"google.golang.org/grpc/attributes"
"google.golang.org/grpc/resolver"
)
type discoveryResolver struct {
w registry.Watcher
cc resolver.ClientConn
ctx context.Context
cancel context.CancelFunc
insecure bool
debugLog bool
selecterKey string
subsetSize int
}
func (r *discoveryResolver) watch() {
for {
select {
case <-r.ctx.Done():
return
default:
}
ins, err := r.w.Next()
if err != nil {
if errors.Is(err, context.Canceled) {
return
}
log.Errorf("[resolver] Failed to watch discovery endpoint: %v", err)
time.Sleep(time.Second)
continue
}
r.update(ins)
}
}
func (r *discoveryResolver) update(ins []*registry.ServiceInstance) {
var (
endpoints = make(map[string]struct{})
filtered = make([]*registry.ServiceInstance, 0, len(ins))
)
for _, in := range ins {
ept, err := endpoint.ParseEndpoint(in.Endpoints, endpoint.Scheme("grpc", !r.insecure))
if err != nil {
log.Errorf("[resolver] Failed to parse discovery endpoint: %v", err)
continue
}
if ept == "" {
continue
}
// filter redundant endpoints
if _, ok := endpoints[ept]; ok {
continue
}
filtered = append(filtered, in)
}
//if r.subsetSize != 0 {
// filtered = subset.Subset(r.selecterKey, filtered, r.subsetSize)
//}
addrs := make([]resolver.Address, 0, len(filtered))
for _, in := range filtered {
ept, _ := endpoint.ParseEndpoint(in.Endpoints, endpoint.Scheme("grpc", !r.insecure))
endpoints[ept] = struct{}{}
addr := resolver.Address{
ServerName: in.Name,
Attributes: parseAttributes(in.Metadata).WithValue("rawServiceInstance", in),
Addr: ept,
}
addrs = append(addrs, addr)
}
if len(addrs) == 0 {
log.Warnf("[resolver] Zero endpoint found,refused to write, instances: %v", ins)
return
}
err := r.cc.UpdateState(resolver.State{Addresses: addrs})
if err != nil {
log.Errorf("[resolver] failed to update state: %s", err)
}
if r.debugLog {
b, _ := json.Marshal(filtered)
log.Infof("[resolver] update instances: %s", b)
}
}
func (r *discoveryResolver) Close() {
r.cancel()
err := r.w.Stop()
if err != nil {
log.Errorf("[resolver] failed to watch top: %s", err)
}
}
func (r *discoveryResolver) ResolveNow(_ resolver.ResolveNowOptions) {}
func parseAttributes(md map[string]string) (a *attributes.Attributes) {
for k, v := range md {
a = a.WithValue(k, v)
}
return a
}
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。