Skip to content

Commit

Permalink
feat: support consul sign config change (#3118)
Browse files Browse the repository at this point in the history
* feat: support consul sign config change

* fix
  • Loading branch information
shenqidebaozi authored Dec 15, 2023
1 parent f753a9f commit 08300d8
Showing 1 changed file with 42 additions and 15 deletions.
57 changes: 42 additions & 15 deletions contrib/config/consul/watcher.go
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@ package consul

import (
"context"
"path/filepath"
"strings"

"github.com/hashicorp/consul/api"
"github.com/hashicorp/consul/api/watch"
Expand All @@ -10,35 +12,60 @@ 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{}) {
if data == nil {
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})
Expand All @@ -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()
}
Expand Down

0 comments on commit 08300d8

Please sign in to comment.