micro-config-vault/watcher.go

107 lines
1.8 KiB
Go
Raw Normal View History

package vault
import (
"encoding/json"
"fmt"
"reflect"
"github.com/hashicorp/vault/api"
"go.unistack.org/micro/v3/config"
"go.unistack.org/micro/v3/util/jitter"
rutil "go.unistack.org/micro/v3/util/reflect"
)
type vaultWatcher struct {
path string
done chan struct{}
vchan chan map[string]interface{}
echan chan error
cli *api.Client
opts config.Options
wopts config.WatchOptions
}
func (w *vaultWatcher) run() {
ticker := jitter.NewTicker(w.wopts.MinInterval, w.wopts.MaxInterval)
defer ticker.Stop()
src := w.opts.Struct
if w.wopts.Struct != nil {
src = w.wopts.Struct
}
for {
select {
case <-w.done:
return
case <-ticker.C:
dst, err := rutil.Zero(src)
if err != nil {
w.echan <- err
return
}
pair, err := w.cli.Logical().Read(w.path)
if err != nil {
w.echan <- err
return
} else if pair == nil || pair.Data == nil {
w.echan <- fmt.Errorf("vault path %s not found", w.path)
return
}
var data []byte
data, err = json.Marshal(pair.Data["data"])
if err == nil {
err = w.opts.Codec.Unmarshal(data, dst)
}
if err != nil {
w.echan <- err
return
}
srcmp, err := rutil.StructFieldsMap(src)
if err != nil {
w.echan <- err
return
}
dstmp, err := rutil.StructFieldsMap(dst)
if err != nil {
w.echan <- err
return
}
for sk, sv := range srcmp {
if reflect.DeepEqual(dstmp[sk], sv) {
delete(dstmp, sk)
}
}
if len(dstmp) > 0 {
w.vchan <- dstmp
src = dst
}
}
}
}
func (w *vaultWatcher) Next() (map[string]interface{}, error) {
select {
case <-w.done:
break
case err := <-w.echan:
return nil, err
case v, ok := <-w.vchan:
if !ok {
break
}
return v, nil
}
return nil, config.ErrWatcherStopped
}
func (w *vaultWatcher) Stop() error {
close(w.done)
return nil
}