mirror of
https://github.com/go-micro/go-micro.git
synced 2025-06-06 22:06:19 +02:00
fix: fix configuration version data competition (#2586)
Co-authored-by: JeffreyBool <zhanggaoyuan@mediatrack.cn>
This commit is contained in:
parent
b988a78bae
commit
b442dbb56b
@ -7,6 +7,7 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"go-micro.dev/v4/config/loader"
|
"go-micro.dev/v4/config/loader"
|
||||||
@ -42,10 +43,14 @@ type watcher struct {
|
|||||||
path []string
|
path []string
|
||||||
value reader.Value
|
value reader.Value
|
||||||
reader reader.Reader
|
reader reader.Reader
|
||||||
version string
|
version atomic.Value
|
||||||
updates chan updateValue
|
updates chan updateValue
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (w *watcher) getVersion() string {
|
||||||
|
return w.version.Load().(string)
|
||||||
|
}
|
||||||
|
|
||||||
func (m *memory) watch(idx int, s source.Source) {
|
func (m *memory) watch(idx int, s source.Source) {
|
||||||
// watches a source for changes
|
// watches a source for changes
|
||||||
watch := func(idx int, s source.Watcher) error {
|
watch := func(idx int, s source.Watcher) error {
|
||||||
@ -169,7 +174,7 @@ func (m *memory) update() {
|
|||||||
m.RUnlock()
|
m.RUnlock()
|
||||||
|
|
||||||
for _, w := range watchers {
|
for _, w := range watchers {
|
||||||
if w.version >= snap.Version {
|
if w.getVersion() >= snap.Version {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -357,8 +362,8 @@ func (m *memory) Watch(path ...string) (loader.Watcher, error) {
|
|||||||
value: value,
|
value: value,
|
||||||
reader: m.opts.Reader,
|
reader: m.opts.Reader,
|
||||||
updates: make(chan updateValue, 1),
|
updates: make(chan updateValue, 1),
|
||||||
version: m.snap.Version,
|
|
||||||
}
|
}
|
||||||
|
w.version.Store(m.snap.Version)
|
||||||
|
|
||||||
e := m.watchers.PushBack(w)
|
e := m.watchers.PushBack(w)
|
||||||
|
|
||||||
@ -392,7 +397,7 @@ func (w *watcher) Next() (*loader.Snapshot, error) {
|
|||||||
|
|
||||||
return &loader.Snapshot{
|
return &loader.Snapshot{
|
||||||
ChangeSet: cs,
|
ChangeSet: cs,
|
||||||
Version: w.version,
|
Version: w.getVersion(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -402,13 +407,13 @@ func (w *watcher) Next() (*loader.Snapshot, error) {
|
|||||||
return nil, errors.New("watcher stopped")
|
return nil, errors.New("watcher stopped")
|
||||||
|
|
||||||
case uv := <-w.updates:
|
case uv := <-w.updates:
|
||||||
if uv.version <= w.version {
|
if uv.version <= w.getVersion() {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
v := uv.value
|
v := uv.value
|
||||||
|
|
||||||
w.version = uv.version
|
w.version.Store(uv.version)
|
||||||
|
|
||||||
if bytes.Equal(w.value.Bytes(), v.Bytes()) {
|
if bytes.Equal(w.value.Bytes(), v.Bytes()) {
|
||||||
continue
|
continue
|
||||||
|
Loading…
x
Reference in New Issue
Block a user