config: minor changes to split config and watcher files

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
Василий Толстов 2021-08-04 13:51:43 +03:00
parent efe215cd60
commit c2f34df493
3 changed files with 82 additions and 77 deletions

View File

@ -7,7 +7,6 @@ import (
"strings"
"github.com/imdario/mergo"
"github.com/unistack-org/micro/v3/util/jitter"
rutil "github.com/unistack-org/micro/v3/util/reflect"
)
@ -292,76 +291,3 @@ func NewConfig(opts ...Option) Config {
}
return &defaultConfig{opts: options}
}
type defaultWatcher struct {
opts Options
wopts WatchOptions
done chan struct{}
vchan chan map[string]interface{}
echan chan error
}
func (w *defaultWatcher) 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 {
err = fillValues(reflect.ValueOf(dst), w.opts.StructTag)
}
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 *defaultWatcher) 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, ErrWatcherStopped
}
func (w *defaultWatcher) Stop() error {
close(w.done)
return nil
}

View File

@ -45,10 +45,8 @@ func TestWatch(t *testing.T) {
go func() {
for {
mp, err := w.Next()
if err != nil && err != config.ErrWatcherStopped {
if err != nil {
t.Fatal(err)
} else if err == config.ErrWatcherStopped {
return
}
if len(mp) != 1 {
t.Fatal(fmt.Errorf("default watcher err: %v", mp))

81
config/watcher.go Normal file
View File

@ -0,0 +1,81 @@
package config
import (
"reflect"
"github.com/unistack-org/micro/v3/util/jitter"
rutil "github.com/unistack-org/micro/v3/util/reflect"
)
type defaultWatcher struct {
opts Options
wopts WatchOptions
done chan struct{}
vchan chan map[string]interface{}
echan chan error
}
func (w *defaultWatcher) 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 {
err = fillValues(reflect.ValueOf(dst), w.opts.StructTag)
}
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 *defaultWatcher) 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, ErrWatcherStopped
}
func (w *defaultWatcher) Stop() error {
close(w.done)
return nil
}