Skip to content

Commit

Permalink
fix: etcd scheme on grpc resolver (#4121)
Browse files Browse the repository at this point in the history
  • Loading branch information
kevwan authored Aug 27, 2024
1 parent 896e1a2 commit 47d13e5
Show file tree
Hide file tree
Showing 2 changed files with 36 additions and 9 deletions.
42 changes: 34 additions & 8 deletions zrpc/resolver/internal/discovbuilder.go
Original file line number Diff line number Diff line change
Expand Up @@ -9,10 +9,41 @@ import (
"google.golang.org/grpc/resolver"
)

type discovBuilder struct{}
type discovBuilder struct {
cc resolver.ClientConn
update func()
}

func (b *discovBuilder) Build(target resolver.Target, cc resolver.ClientConn, _ resolver.BuildOptions) (
resolver.Resolver, error) {
b.cc = cc
if err := b.updateState(target); err != nil {
return nil, err
}

return &nopResolver{cc: cc}, nil
}

func (b *discovBuilder) Scheme() string {
return DiscovScheme
}

func (b *discovBuilder) updateState(target resolver.Target) error {
if b.update == nil {
update, err := b.buildEndpointsUpdater(target)
if err != nil {
return err
}

b.update = update
}

b.update()

return nil
}

func (b *discovBuilder) buildEndpointsUpdater(target resolver.Target) (func(), error) {
hosts := strings.FieldsFunc(targets.GetAuthority(target), func(r rune) bool {
return r == EndpointSepChar
})
Expand All @@ -29,18 +60,13 @@ func (b *discovBuilder) Build(target resolver.Target, cc resolver.ClientConn, _
Addr: val,
})
}
if err := cc.UpdateState(resolver.State{
if err := b.cc.UpdateState(resolver.State{
Addresses: addrs,
}); err != nil {
logx.Error(err)
}
}
sub.AddListener(update)
update()

return &nopResolver{cc: cc}, nil
}

func (b *discovBuilder) Scheme() string {
return DiscovScheme
return update, nil
}
3 changes: 2 additions & 1 deletion zrpc/resolver/internal/kubebuilder.go
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,8 @@ type kubeResolver struct {
stopCh chan struct{}
}

func (r *kubeResolver) ResolveNow(_ resolver.ResolveNowOptions) {}
func (r *kubeResolver) ResolveNow(_ resolver.ResolveNowOptions) {
}

func (r *kubeResolver) start() {
threading.GoSafe(func() {
Expand Down

0 comments on commit 47d13e5

Please sign in to comment.