feat: support consul sign config change (#3118)

* feat: support consul sign config change

* fix

contrib/config/consul/watcher.go

@@ -2,6 +2,8 @@ package consul
 
 import (
 	"context"
+	"path/filepath"
+	"strings"
 
 	"github.com/hashicorp/consul/api"
 	"github.com/hashicorp/consul/api/watch"
@@ -10,12 +12,12 @@ import (
 )
 
 type watcher struct {
-	source *source
-	ch     chan interface{}
-	wp     *watch.Plan
-
-	ctx    context.Context
-	cancel context.CancelFunc
+	source          *source
+	ch              chan []*config.KeyValue
+	wp              *watch.Plan
+	fileModifyIndex map[string]uint64
+	ctx             context.Context
+	cancel          context.CancelFunc
 }
 
 func (w *watcher) handle(_ uint64, data interface{}) {
@@ -23,22 +25,47 @@ func (w *watcher) handle(_ uint64, data interface{}) {
 		return
 	}
 
-	_, ok := data.(api.KVPairs)
+	kv, ok := data.(api.KVPairs)
 	if !ok {
 		return
 	}
 
-	w.ch <- struct{}{}
+	pathPrefix := w.source.options.path
+	if !strings.HasSuffix(w.source.options.path, "/") {
+		pathPrefix = pathPrefix + "/"
+	}
+	kvs := make([]*config.KeyValue, 0, len(kv))
+	for _, item := range kv {
+		if index, ok := w.fileModifyIndex[item.Key]; ok && item.ModifyIndex == index {
+			continue
+		}
+		k := strings.TrimPrefix(item.Key, pathPrefix)
+		if k == "" {
+			continue
+		}
+		kvs = append(kvs, &config.KeyValue{
+			Key:    k,
+			Value:  item.Value,
+			Format: strings.TrimPrefix(filepath.Ext(k), "."),
+		})
+		w.fileModifyIndex[item.Key] = item.ModifyIndex
+	}
+
+	if len(kvs) == 0 {
+		return
+	}
+
+	w.ch <- kvs
 }
 
 func newWatcher(s *source) (*watcher, error) {
 	ctx, cancel := context.WithCancel(context.Background())
 	w := &watcher{
-		source: s,
-		ch:     make(chan interface{}),
-
-		ctx:    ctx,
-		cancel: cancel,
+		source:          s,
+		ch:              make(chan []*config.KeyValue),
+		fileModifyIndex: make(map[string]uint64),
+		ctx:             ctx,
+		cancel:          cancel,
 	}
 
 	wp, err := watch.Parse(map[string]interface{}{"type": "keyprefix", "prefix": s.options.path})
@@ -62,8 +89,8 @@ func newWatcher(s *source) (*watcher, error) {
 
 func (w *watcher) Next() ([]*config.KeyValue, error) {
 	select {
-	case <-w.ch:
-		return w.source.Load()
+	case kv := <-w.ch:
+		return kv, nil
 	case <-w.ctx.Done():
 		return nil, w.ctx.Err()
 	}