Skip to content

Commit

Permalink
chore: coding style (#3796)
Browse files Browse the repository at this point in the history
  • Loading branch information
kevwan committed Dec 17, 2023
1 parent 48625fa commit 27c4908
Showing 1 changed file with 15 additions and 12 deletions.
27 changes: 15 additions & 12 deletions core/discov/internal/registry.go
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@ import (
"github.com/zeromicro/go-zero/core/logx"
"github.com/zeromicro/go-zero/core/syncx"
"github.com/zeromicro/go-zero/core/threading"
v3rpc "go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
clientv3 "go.etcd.io/etcd/client/v3"
)

Expand All @@ -24,6 +24,7 @@ var (
clusters: make(map[string]*cluster),
}
connManager = syncx.NewResourceManager()
errClosed = errors.New("etcd monitor chan has been closed")
)

// A Registry is a registry that manages the etcd client connections.
Expand Down Expand Up @@ -294,36 +295,38 @@ func (c *cluster) watch(cli EtcdClient, key string, rev int64) {
if err == nil {
return
}
if rev != 0 && errors.Is(err, v3rpc.ErrCompacted) {
logx.Errorf("etcd watch stream has been compacted, try to reload, rev %v", rev)

if rev != 0 && errors.Is(err, rpctypes.ErrCompacted) {
logx.Errorf("etcd watch stream has been compacted, try to reload, rev %d", rev)
rev = c.load(cli, key)
}

// log the error and retry
logx.Error(err)
}
}

func (c *cluster) watchStream(cli EtcdClient, key string, rev int64) error {
var rch clientv3.WatchChan
if rev != 0 {
rch = cli.Watch(clientv3.WithRequireLeader(c.context(cli)), makeKeyPrefix(key), clientv3.WithPrefix(),
clientv3.WithRev(rev+1))
rch = cli.Watch(clientv3.WithRequireLeader(c.context(cli)), makeKeyPrefix(key),
clientv3.WithPrefix(), clientv3.WithRev(rev+1))
} else {
rch = cli.Watch(clientv3.WithRequireLeader(c.context(cli)), makeKeyPrefix(key), clientv3.WithPrefix())
rch = cli.Watch(clientv3.WithRequireLeader(c.context(cli)), makeKeyPrefix(key),
clientv3.WithPrefix())
}

for {
select {
case wresp, ok := <-rch:
if !ok {
logx.Error("etcd monitor chan has been closed")
return errors.New("etcd monitor chan has been closed")
return errClosed
}
if wresp.Canceled {
logx.Errorf("etcd monitor chan has been canceled, error: %v", wresp.Err())
return wresp.Err()
return fmt.Errorf("etcd monitor chan has been canceled, error: %w", wresp.Err())
}
if wresp.Err() != nil {
logx.Error(fmt.Sprintf("etcd monitor chan error: %v", wresp.Err()))
return wresp.Err()
return fmt.Errorf("etcd monitor chan error: %w", wresp.Err())
}

c.handleWatchEvents(key, wresp.Events)
Expand Down

0 comments on commit 27c4908

Please sign in to comment.