1 Star 0 Fork 0

llakcs / agile-go

加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
克隆/下载
resolver.go 2.52 KB
一键复制 编辑 原始数据 按行查看 历史
llakcs 提交于 2024-01-31 16:54 . 第一次提交
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
}
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化
1
https://gitee.com/llakcs/agile-go.git
git@gitee.com:llakcs/agile-go.git
llakcs
agile-go
agile-go
v1.2.0

搜索帮助

344bd9b3 5694891 D2dac590 5694891