feat(config): add ErrorWatcherStopped and return on watcher.Next() after Watcher.Stop() (#2092)

* feat(config): add ErrorWatcherStopped and return watcher.Next() after watcher.Stop()

* fix: remove unreachable code

* fix: return err when etcd WatcheResponse hold error

config/config.go

@@ -20,6 +20,8 @@ var (
 	ErrNotFound = errors.New("key not found")
 	// ErrTypeAssert is type assert error.
 	ErrTypeAssert = errors.New("type assert error")
+	// ErrWatcherStopped means watcher is stopped.
+	ErrWatcherStopped = errors.New("warcher stopped")
 
 	_ Config = (*config)(nil)
 )

contrib/config/apollo/watcher.go

@@ -1,8 +1,6 @@
 package apollo
 
 import (
-	"context"
-
 	"github.com/go-kratos/kratos/v2/config"
 	"github.com/go-kratos/kratos/v2/log"
 
@@ -62,7 +60,7 @@ func newWatcher(a *apollo) (config.Watcher, error) {
 func (w *watcher) Next() ([]*config.KeyValue, error) {
 	kv, ok := <-w.out
 	if !ok {
-		return nil, context.Canceled
+		return nil, config.ErrWatcherStopped
 	}
 	return kv, nil
 }

contrib/config/consul/watcher.go

@@ -54,13 +54,10 @@ func newWatcher(s *source) (*watcher, error) {
 
 func (w *watcher) Next() ([]*config.KeyValue, error) {
 	select {
-	case _, ok := <-w.ch:
+	case <-w.ch:
-		if !ok {
-			return nil, nil
-		}
 		return w.source.Load()
 	case <-w.closeChan:
-		return nil, nil
+		return nil, config.ErrWatcherStopped
 	}
 }
 

contrib/config/etcd/watcher.go

@@ -28,13 +28,13 @@ func newWatcher(s *source) *watcher {
 
 func (s *watcher) Next() ([]*config.KeyValue, error) {
 	select {
-	case _, ok := <-s.ch:
+	case resp := <-s.ch:
-		if !ok {
+		if resp.Err() != nil {
-			return nil, nil
+			return nil, resp.Err()
 		}
 		return s.source.Load()
 	case <-s.closeChan:
-		return nil, nil
+		return nil, config.ErrWatcherStopped
 	}
 }
 

contrib/config/nacos/config.go

@@ -106,7 +106,7 @@ func (c *Config) Watch() (config.Watcher, error) {
 	err := c.client.ListenConfig(vo.ConfigParam{
 		DataId: c.opts.dataID,
 		Group:  c.opts.group,
-		OnChange: func(namespace, group, dataId, data string) {
+		OnChange: func(_, group, dataId, data string) {
 			if dataId == watcher.dataID && group == watcher.group {
 				watcher.content <- data
 			}
@@ -144,8 +144,8 @@ func newWatcher(ctx context.Context, dataID string, group string, cancelListenCo
 
 func (w *Watcher) Next() ([]*config.KeyValue, error) {
 	select {
-	case <-w.Context.Done():
+	case <-w.Done():
-		return nil, nil
+		return nil, config.ErrWatcherStopped
 	case content := <-w.content:
 		k := w.dataID
 		return []*config.KeyValue{