diff --git a/registry/consul/listener.go b/registry/consul/listener.go
index cf3888dd16ee4a9b504664383be69ca1faf4d842..b1598345032bddd7358c2263e5fd71b3d4e5545d 100644
--- a/registry/consul/listener.go
+++ b/registry/consul/listener.go
@@ -197,7 +197,7 @@ func (l *consulListener) Next() (*registry.ServiceEvent, error) {
}
}
-// Close closes this listener
+// Close the listener.
func (l *consulListener) Close() {
close(l.done)
l.plan.Stop()
diff --git a/remoting/etcdv3/client.go b/remoting/etcdv3/client.go
index 7632a7cd042d36db5e02280a14224b83e8736e6c..4e7436e445f652d8fe1db2991c87303653beb9ec 100644
--- a/remoting/etcdv3/client.go
+++ b/remoting/etcdv3/client.go
@@ -35,7 +35,7 @@ import (
)
const (
- // ConnDelay connection dalay
+ // ConnDelay connection delay
ConnDelay = 3
// MaxFailTimes max failure times
MaxFailTimes = 15
@@ -93,7 +93,6 @@ func WithHeartbeat(heartbeat int) Option {
// ValidateClient validates client and sets options
func ValidateClient(container clientFacade, opts ...Option) error {
-
options := &Options{
heartbeat: 1, // default heartbeat
}
@@ -118,7 +117,6 @@ func ValidateClient(container clientFacade, opts ...Option) error {
// Client lose connection with etcd server
if container.Client().rawClient == nil {
-
newClient, err := NewClient(options.name, options.endpoints, options.timeout, options.heartbeat)
if err != nil {
logger.Warnf("new etcd client (name{%s}, etcd addresses{%v}, timeout{%d}) = error{%v}",
@@ -136,7 +134,6 @@ func NewServiceDiscoveryClient(opts ...Option) *Client {
options := &Options{
heartbeat: 1, // default heartbeat
}
-
for _, opt := range opts {
opt(options)
}
@@ -145,9 +142,7 @@ func NewServiceDiscoveryClient(opts ...Option) *Client {
if err != nil {
logger.Errorf("new etcd client (name{%s}, etcd addresses{%v}, timeout{%d}) = error{%v}",
options.name, options.endpoints, options.timeout, err)
- return nil
}
-
return newClient
}
@@ -171,7 +166,6 @@ type Client struct {
// nolint
func NewClient(name string, endpoints []string, timeout time.Duration, heartbeat int) (*Client, error) {
-
ctx, cancel := context.WithCancel(context.Background())
rawClient, err := clientv3.New(clientv3.Config{
Context: ctx,
@@ -184,7 +178,6 @@ func NewClient(name string, endpoints []string, timeout time.Duration, heartbeat
}
c := &Client{
-
name: name,
timeout: timeout,
endpoints: endpoints,
@@ -205,7 +198,6 @@ func NewClient(name string, endpoints []string, timeout time.Duration, heartbeat
// NOTICE: need to get the lock before calling this method
func (c *Client) clean() {
-
// close raw client
c.rawClient.Close()
@@ -217,7 +209,6 @@ func (c *Client) clean() {
}
func (c *Client) stop() bool {
-
select {
case <-c.exit:
return true
@@ -229,7 +220,6 @@ func (c *Client) stop() bool {
// nolint
func (c *Client) Close() {
-
if c == nil {
return
}
@@ -241,15 +231,14 @@ func (c *Client) Close() {
c.Wait.Wait()
c.lock.Lock()
+ defer c.lock.Unlock()
if c.rawClient != nil {
c.clean()
}
- c.lock.Unlock()
logger.Warnf("etcd client{name:%s, endpoints:%s} exit now.", c.name, c.endpoints)
}
func (c *Client) maintenanceStatus() error {
-
s, err := concurrency.NewSession(c.rawClient, concurrency.WithTTL(c.heartbeat))
if err != nil {
return perrors.WithMessage(err, "new session with server")
@@ -262,7 +251,6 @@ func (c *Client) maintenanceStatus() error {
}
func (c *Client) maintenanceStatusLoop(s *concurrency.Session) {
-
defer func() {
c.Wait.Done()
logger.Infof("etcd client {endpoints:%v, name:%s} maintenance goroutine game over.", c.endpoints, c.name)
@@ -288,7 +276,6 @@ func (c *Client) maintenanceStatusLoop(s *concurrency.Session) {
// if k not exist will put k/v in etcd, otherwise return nil
func (c *Client) put(k string, v string, opts ...clientv3.OpOption) error {
-
c.lock.RLock()
defer c.lock.RUnlock()
@@ -300,17 +287,12 @@ func (c *Client) put(k string, v string, opts ...clientv3.OpOption) error {
If(clientv3.Compare(clientv3.Version(k), "<", 1)).
Then(clientv3.OpPut(k, v, opts...)).
Commit()
- if err != nil {
- return err
-
- }
- return nil
+ return err
}
// if k not exist will put k/v in etcd
// if k is already exist in etcd, replace it
func (c *Client) update(k string, v string, opts ...clientv3.OpOption) error {
-
c.lock.RLock()
defer c.lock.RUnlock()
@@ -322,15 +304,10 @@ func (c *Client) update(k string, v string, opts ...clientv3.OpOption) error {
If(clientv3.Compare(clientv3.Version(k), "!=", -1)).
Then(clientv3.OpPut(k, v, opts...)).
Commit()
- if err != nil {
- return err
-
- }
- return nil
+ return err
}
func (c *Client) delete(k string) error {
-
c.lock.RLock()
defer c.lock.RUnlock()
@@ -339,15 +316,10 @@ func (c *Client) delete(k string) error {
}
_, err := c.rawClient.Delete(c.ctx, k)
- if err != nil {
- return err
-
- }
- return nil
+ return err
}
func (c *Client) get(k string) (string, error) {
-
c.lock.RLock()
defer c.lock.RUnlock()
@@ -369,7 +341,6 @@ func (c *Client) get(k string) (string, error) {
// nolint
func (c *Client) CleanKV() error {
-
c.lock.RLock()
defer c.lock.RUnlock()
@@ -378,15 +349,10 @@ func (c *Client) CleanKV() error {
}
_, err := c.rawClient.Delete(c.ctx, "", clientv3.WithPrefix())
- if err != nil {
- return err
- }
-
- return nil
+ return err
}
func (c *Client) getChildren(k string) ([]string, []string, error) {
-
c.lock.RLock()
defer c.lock.RUnlock()
@@ -403,21 +369,16 @@ func (c *Client) getChildren(k string) ([]string, []string, error) {
return nil, nil, ErrKVPairNotFound
}
- var (
- kList []string
- vList []string
- )
-
+ kList := make([]string, 0, len(resp.Kvs))
+ vList := make([]string, 0, len(resp.Kvs))
for _, kv := range resp.Kvs {
kList = append(kList, string(kv.Key))
vList = append(vList, string(kv.Value))
}
-
return kList, vList, nil
}
func (c *Client) watchWithPrefix(prefix string) (clientv3.WatchChan, error) {
-
c.lock.RLock()
defer c.lock.RUnlock()
@@ -429,7 +390,6 @@ func (c *Client) watchWithPrefix(prefix string) (clientv3.WatchChan, error) {
}
func (c *Client) watch(k string) (clientv3.WatchChan, error) {
-
c.lock.RLock()
defer c.lock.RUnlock()
@@ -441,7 +401,6 @@ func (c *Client) watch(k string) (clientv3.WatchChan, error) {
}
func (c *Client) keepAliveKV(k string, v string) error {
-
c.lock.RLock()
defer c.lock.RUnlock()
@@ -457,14 +416,15 @@ func (c *Client) keepAliveKV(k string, v string) error {
keepAlive, err := c.rawClient.KeepAlive(c.ctx, lease.ID)
if err != nil || keepAlive == nil {
c.rawClient.Revoke(c.ctx, lease.ID)
- return perrors.WithMessage(err, "keep alive lease")
+ if err != nil {
+ return perrors.WithMessage(err, "keep alive lease")
+ } else {
+ return perrors.New("keep alive lease")
+ }
}
_, err = c.rawClient.Put(c.ctx, k, v, clientv3.WithLease(lease.ID))
- if err != nil {
- return perrors.WithMessage(err, "put k/v with lease")
- }
- return nil
+ return perrors.WithMessage(err, "put k/v with lease")
}
// nolint
@@ -481,92 +441,54 @@ func (c *Client) Valid() bool {
}
c.lock.RLock()
- if c.rawClient == nil {
- c.lock.RUnlock()
- return false
- }
- c.lock.RUnlock()
- return true
+ defer c.lock.RUnlock()
+ return c.rawClient != nil
}
// nolint
func (c *Client) Create(k string, v string) error {
-
err := c.put(k, v)
- if err != nil {
- return perrors.WithMessagef(err, "put k/v (key: %s value %s)", k, v)
- }
- return nil
+ return perrors.WithMessagef(err, "put k/v (key: %s value %s)", k, v)
}
// Update key value ...
func (c *Client) Update(k, v string) error {
err := c.update(k, v)
- if err != nil {
- return perrors.WithMessagef(err, "Update k/v (key: %s value %s)", k, v)
- }
- return nil
+ return perrors.WithMessagef(err, "Update k/v (key: %s value %s)", k, v)
}
// nolint
func (c *Client) Delete(k string) error {
-
err := c.delete(k)
- if err != nil {
- return perrors.WithMessagef(err, "delete k/v (key %s)", k)
- }
-
- return nil
+ return perrors.WithMessagef(err, "delete k/v (key %s)", k)
}
// RegisterTemp registers a temporary node
func (c *Client) RegisterTemp(k, v string) error {
-
err := c.keepAliveKV(k, v)
- if err != nil {
- return perrors.WithMessagef(err, "keepalive kv (key %s)", k)
- }
-
- return nil
+ return perrors.WithMessagef(err, "keepalive kv (key %s)", k)
}
// GetChildrenKVList gets children kv list by @k
func (c *Client) GetChildrenKVList(k string) ([]string, []string, error) {
-
kList, vList, err := c.getChildren(k)
- if err != nil {
- return nil, nil, perrors.WithMessagef(err, "get key children (key %s)", k)
- }
- return kList, vList, nil
+ return kList, vList, perrors.WithMessagef(err, "get key children (key %s)", k)
}
// Get gets value by @k
func (c *Client) Get(k string) (string, error) {
-
v, err := c.get(k)
- if err != nil {
- return "", perrors.WithMessagef(err, "get key value (key %s)", k)
- }
-
- return v, nil
+ return v, perrors.WithMessagef(err, "get key value (key %s)", k)
}
// Watch watches on spec key
func (c *Client) Watch(k string) (clientv3.WatchChan, error) {
-
wc, err := c.watch(k)
- if err != nil {
- return nil, perrors.WithMessagef(err, "watch prefix (key %s)", k)
- }
- return wc, nil
+ return wc, perrors.WithMessagef(err, "watch prefix (key %s)", k)
}
// WatchWithPrefix watches on spec prefix
func (c *Client) WatchWithPrefix(prefix string) (clientv3.WatchChan, error) {
-
wc, err := c.watchWithPrefix(prefix)
- if err != nil {
- return nil, perrors.WithMessagef(err, "watch prefix (key %s)", prefix)
- }
- return wc, nil
+ return wc, perrors.WithMessagef(err, "watch prefix (key %s)", prefix)
}