You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
97 lines
2.0 KiB
97 lines
2.0 KiB
package apollo
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
|
|
"github.com/go-kratos/kratos/v2/config"
|
|
"github.com/go-kratos/kratos/v2/encoding"
|
|
"github.com/go-kratos/kratos/v2/log"
|
|
|
|
"github.com/apolloconfig/agollo/v4/storage"
|
|
)
|
|
|
|
type watcher struct {
|
|
out <-chan []*config.KeyValue
|
|
cancelFn func()
|
|
}
|
|
|
|
type customChangeListener struct {
|
|
in chan<- []*config.KeyValue
|
|
logger log.Logger
|
|
}
|
|
|
|
func (c *customChangeListener) onChange(
|
|
namespace string, changes map[string]*storage.ConfigChange) []*config.KeyValue {
|
|
|
|
kv := make([]*config.KeyValue, 0, 2)
|
|
next := make(map[string]interface{})
|
|
|
|
for key, change := range changes {
|
|
resolve(genKey(namespace, key), change.NewValue, next)
|
|
}
|
|
|
|
f := format(namespace)
|
|
codec := encoding.GetCodec(f)
|
|
val, err := codec.Marshal(next)
|
|
if err != nil {
|
|
_ = c.logger.Log(log.LevelWarn,
|
|
"msg",
|
|
fmt.Sprintf("apollo could not handle namespace %s: %v", namespace, err),
|
|
)
|
|
return nil
|
|
}
|
|
kv = append(kv, &config.KeyValue{
|
|
Key: namespace,
|
|
Value: val,
|
|
Format: f,
|
|
})
|
|
|
|
return kv
|
|
}
|
|
|
|
func (c *customChangeListener) OnChange(changeEvent *storage.ChangeEvent) {
|
|
change := c.onChange(changeEvent.Namespace, changeEvent.Changes)
|
|
if len(change) == 0 {
|
|
return
|
|
}
|
|
|
|
c.in <- change
|
|
}
|
|
|
|
func (c *customChangeListener) OnNewestChange(changeEvent *storage.FullChangeEvent) {}
|
|
|
|
func newWatcher(a *apollo, logger log.Logger) (config.Watcher, error) {
|
|
if logger == nil {
|
|
logger = log.DefaultLogger
|
|
}
|
|
|
|
changeCh := make(chan []*config.KeyValue)
|
|
listener := &customChangeListener{in: changeCh, logger: logger}
|
|
a.client.AddChangeListener(listener)
|
|
|
|
return &watcher{
|
|
out: changeCh,
|
|
cancelFn: func() {
|
|
a.client.RemoveChangeListener(listener)
|
|
close(changeCh)
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
// Next will be blocked until the Stop method is called
|
|
func (w *watcher) Next() ([]*config.KeyValue, error) {
|
|
kv, ok := <-w.out
|
|
if !ok {
|
|
return nil, context.Canceled
|
|
}
|
|
return kv, nil
|
|
}
|
|
|
|
func (w *watcher) Stop() error {
|
|
if w.cancelFn != nil {
|
|
w.cancelFn()
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|