fix:config groutine(watch) leak (#1327)

* fix:config goroutine(watch) Leak

config/config.go

@@ -1,6 +1,7 @@
 package config
 
 import (
+	"context"
 	"errors"
 	"reflect"
 	"sync"
@@ -65,6 +66,10 @@ func New(opts ...Option) Config {
 func (c *config) watch(w Watcher) {
 	for {
 		kvs, err := w.Next()
+		if errors.Is(err, context.Canceled) {
+			c.log.Infof("watcher's ctx cancel : %v", err)
+			return
+		}
 		if err != nil {
 			time.Sleep(time.Second)
 			c.log.Errorf("failed to watch next config: %v", err)

config/env/watcher.go

@@ -1,24 +1,37 @@
 package env
 
 import (
+	"context"
+
 	"github.com/go-kratos/kratos/v2/config"
 )
 
 type watcher struct {
 	exit chan struct{}
+
+	ctx    context.Context
+	cancel context.CancelFunc
 }
 
+var _ config.Watcher = (*watcher)(nil)
+
 func NewWatcher() (config.Watcher, error) {
-	return &watcher{exit: make(chan struct{})}, nil
+	ctx, cancel := context.WithCancel(context.Background())
+	return &watcher{exit: make(chan struct{}), ctx: ctx, cancel: cancel}, nil
 }
 
 // Next will be blocked until the Stop method is called
 func (w *watcher) Next() ([]*config.KeyValue, error) {
-	<-w.exit
+	select {
+	case <-w.ctx.Done():
+		return nil, w.ctx.Err()
+	case <-w.exit:
 		return nil, nil
+	}
 }
 
 func (w *watcher) Stop() error {
 	close(w.exit)
+	w.cancel()
 	return nil
 }

config/file/watcher.go

@@ -1,6 +1,7 @@
 package file
 
 import (
+	"context"
 	"os"
 	"path/filepath"
 
@@ -11,8 +12,13 @@ import (
 type watcher struct {
 	f  *file
 	fw *fsnotify.Watcher
+
+	ctx    context.Context
+	cancel context.CancelFunc
 }
 
+var _ config.Watcher = (*watcher)(nil)
+
 func newWatcher(f *file) (config.Watcher, error) {
 	fw, err := fsnotify.NewWatcher()
 	if err != nil {
@@ -21,11 +27,14 @@ func newWatcher(f *file) (config.Watcher, error) {
 	if err := fw.Add(f.path); err != nil {
 		return nil, err
 	}
-	return &watcher{f: f, fw: fw}, nil
+	ctx, cancel := context.WithCancel(context.Background())
+	return &watcher{f: f, fw: fw, ctx: ctx, cancel: cancel}, nil
 }
 
 func (w *watcher) Next() ([]*config.KeyValue, error) {
 	select {
+	case <-w.ctx.Done():
+		return nil, w.ctx.Err()
 	case event := <-w.fw.Events:
 		if event.Op == fsnotify.Rename {
 			if _, err := os.Stat(event.Name); err == nil || os.IsExist(err) {
@@ -53,5 +62,6 @@ func (w *watcher) Next() ([]*config.KeyValue, error) {
 }
 
 func (w *watcher) Stop() error {
+	w.cancel()
 	return w.fw.Close()
 }