package runtime import ( "context" "errors" "reflect" "sync" "time" ) var ErrNotFound = errors.New("key not found") type Observer func(string, Value) type Config interface { Load() error Scan(any) error Value(string) Value Watch(string, Observer) error Close() error } type config struct { opts options reader Reader cached sync.Map observers sync.Map watchers []Watcher closeOnce sync.Once } func New(opts ...Option) Config { o := options{ decoder: defaultDecoder, resolver: defaultResolver, merge: mergeMap, } for _, opt := range opts { opt(&o) } return &config{ opts: o, reader: newReader(o), } } func (c *config) Load() error { for _, src := range c.opts.sources { kvs, err := src.Load() if err != nil { return err } if err = c.reader.Merge(kvs...); err != nil { return err } w, err := src.Watch() if err != nil { return err } c.watchers = append(c.watchers, w) go c.watch(w) } return c.reader.Resolve() } func (c *config) Value(key string) Value { if v, ok := c.cached.Load(key); ok { return v.(Value) } if v, ok := c.reader.Value(key); ok { c.cached.Store(key, v) return v } return &errValue{err: ErrNotFound} } func (c *config) Scan(v any) error { data, err := c.reader.Source() if err != nil { return err } return unmarshalJSON(data, v) } func (c *config) Watch(key string, o Observer) error { if v := c.Value(key); v.Load() == nil { return ErrNotFound } c.observers.Store(key, o) return nil } func (c *config) Close() error { var closeErr error c.closeOnce.Do(func() { for _, w := range c.watchers { if err := w.Stop(); err != nil && closeErr == nil { closeErr = err } } }) return closeErr } func (c *config) watch(w Watcher) { for { kvs, err := w.Next() if err != nil { if errors.Is(err, context.Canceled) { return } time.Sleep(time.Second) continue } if err := c.reader.Merge(kvs...); err != nil { c.notifyRootObserver() continue } if err := c.reader.Resolve(); err != nil { c.notifyRootObserver() continue } c.notifyObservers() } } func (c *config) notifyRootObserver() { if observer, ok := c.observers.Load(""); ok { observer.(Observer)("", c.Value("")) } } func (c *config) notifyObservers() { if observer, ok := c.observers.Load(""); ok { next, ok := c.reader.Value("") if !ok { return } current := c.Value("") current.Store(next.Load()) observer.(Observer)("", current) } c.cached.Range(func(key, value any) bool { path := key.(string) if path == "" { return true } previous := value.(Value) next, ok := c.reader.Value(path) if !ok { return true } if reflect.TypeOf(next.Load()) == reflect.TypeOf(previous.Load()) && !reflect.DeepEqual(next.Load(), previous.Load()) { previous.Store(next.Load()) if observer, ok := c.observers.Load(path); ok { observer.(Observer)(path, previous) } } return true }) } func mergeMap(dst, src any) error { dstMap, ok := dst.(*map[string]any) if !ok { return errors.New("merge destination must be *map[string]any") } srcMap, ok := src.(map[string]any) if !ok { return errors.New("merge source must be map[string]any") } deepMerge(*dstMap, srcMap) return nil } func deepMerge(dst, src map[string]any) { for key, srcValue := range src { if srcChild, ok := srcValue.(map[string]any); ok { if dstChild, ok := dst[key].(map[string]any); ok { deepMerge(dstChild, srcChild) continue } } dst[key] = srcValue } }