feat(os/gcfg): Add file watcher with custom callback support

contrib/config/apollo/apollo.go

@@ -22,6 +22,12 @@ import (
 	"github.com/gogf/gf/v2/util/gconv"
 )
 
+var (
+	// Compile-time checking for interface implementation.
+	_ gcfg.Adapter        = (*Client)(nil)
+	_ gcfg.WatcherAdapter = (*Client)(nil)
+)
+
 // Config is the configuration object for apollo client.
 type Config struct {
 	AppID             string `v:"required"` // See apolloConfig.Config.
@@ -38,9 +44,10 @@ type Config struct {
 
 // Client implements gcfg.Adapter implementing using apollo service.
 type Client struct {
-	config Config        // Config object when created.
-	client agollo.Client // Apollo client.
-	value  *g.Var        // Configmap content cached. It is `*gjson.Json` value internally.
+	config   Config                // Config object when created.
+	client   agollo.Client         // Apollo client.
+	value    *g.Var                // Configmap content cached. It is `*gjson.Json` value internally.
+	watchers *gcfg.WatcherRegistry // Watchers for watching file changes.
 }
 
 // New creates and returns gcfg.Adapter implementing using apollo service.
@@ -54,8 +61,9 @@ func New(ctx context.Context, config Config) (adapter gcfg.Adapter, err error) {
 		config.NamespaceName = storage.GetDefaultNamespace()
 	}
 	client := &Client{
-		config: config,
-		value:  g.NewVar(nil, true),
+		config:   config,
+		value:    g.NewVar(nil, true),
+		watchers: gcfg.NewWatcherRegistry(),
 	}
 	// Apollo client.
 	client.client, err = agollo.StartWithConfig(func() (*apolloConfig.AppConfig, error) {
@@ -89,7 +97,7 @@ func (c *Client) Available(ctx context.Context, resource ...string) (ok bool) {
 	if len(resource) == 0 && !c.value.IsNil() {
 		return true
 	}
-	var namespace = c.config.NamespaceName
+	namespace := c.config.NamespaceName
 	if len(resource) > 0 {
 		namespace = resource[0]
 	}
@@ -132,18 +140,46 @@ func (c *Client) OnNewestChange(event *storage.FullChangeEvent) {
 }
 
 func (c *Client) updateLocalValue(ctx context.Context) (err error) {
-	var j = gjson.New(nil)
+	j := gjson.New(nil)
+	content := gjson.New(nil, true)
 	cache := c.client.GetConfigCache(c.config.NamespaceName)
 	cache.Range(func(key, value any) bool {
 		err = j.Set(gconv.String(key), value)
 		if err != nil {
 			return false
 		}
+		err = content.Set(gconv.String(key), value)
+		if err != nil {
+			return false
+		}
 		return true
 	})
 	cache.Clear()
 	if err == nil {
 		c.value.Set(j)
+		adapterCtx := NewAdapterCtx(ctx).WithOperation(gcfg.OperationUpdate).WithNamespace(c.config.NamespaceName).
+			WithAppId(c.config.AppID).WithCluster(c.config.Cluster).WithContent(content)
+		c.notifyWatchers(adapterCtx.Ctx)
 	}
 	return
 }
+
+// AddWatcher adds a watcher for the specified configuration file.
+func (c *Client) AddWatcher(name string, f func(ctx context.Context)) {
+	c.watchers.Add(name, f)
+}
+
+// RemoveWatcher removes the watcher for the specified configuration file.
+func (c *Client) RemoveWatcher(name string) {
+	c.watchers.Remove(name)
+}
+
+// GetWatcherNames returns all watcher names.
+func (c *Client) GetWatcherNames() []string {
+	return c.watchers.GetNames()
+}
+
+// notifyWatchers notifies all watchers.
+func (c *Client) notifyWatchers(ctx context.Context) {
+	c.watchers.Notify(ctx)
+}

contrib/config/apollo/apollo_adapter_ctx.go

@@ -0,0 +1,132 @@
+// Copyright GoFrame Author(https://goframe.org). All Rights Reserved.
+//
+// This Source Code Form is subject to the terms of the MIT License.
+// If a copy of the MIT was not distributed with this file,
+// You can obtain one at https://github.com/gogf/gf.
+
+// Package apollo implements gcfg.Adapter using apollo service.
+package apollo
+
+import (
+	"context"
+
+	"github.com/gogf/gf/v2/encoding/gjson"
+	"github.com/gogf/gf/v2/os/gcfg"
+	"github.com/gogf/gf/v2/os/gctx"
+)
+
+const (
+	// ContextKeyNamespace is the context key for namespace
+	ContextKeyNamespace gctx.StrKey = "namespace"
+	// ContextKeyAppId is the context key for appId
+	ContextKeyAppId gctx.StrKey = "appId"
+	// ContextKeyCluster is the context key for cluster
+	ContextKeyCluster gctx.StrKey = "cluster"
+)
+
+// ApolloAdapterCtx is the context adapter for Apollo configuration
+type ApolloAdapterCtx struct {
+	Ctx context.Context
+}
+
+// NewAdapterCtxWithCtx creates and returns a new ApolloAdapterCtx with the given context.
+func NewAdapterCtxWithCtx(ctx context.Context) *ApolloAdapterCtx {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	return &ApolloAdapterCtx{Ctx: ctx}
+}
+
+// NewAdapterCtx creates and returns a new ApolloAdapterCtx.
+// If context is provided, it will be used; otherwise, a background context is created.
+func NewAdapterCtx(ctx ...context.Context) *ApolloAdapterCtx {
+	if len(ctx) > 0 {
+		return NewAdapterCtxWithCtx(ctx[0])
+	}
+	return NewAdapterCtxWithCtx(context.Background())
+}
+
+// GetAdapterCtx creates a new ApolloAdapterCtx with the given context
+func GetAdapterCtx(ctx context.Context) *ApolloAdapterCtx {
+	return NewAdapterCtxWithCtx(ctx)
+}
+
+// WithOperation sets the operation in the context
+func (a *ApolloAdapterCtx) WithOperation(operation gcfg.OperationType) *ApolloAdapterCtx {
+	a.Ctx = context.WithValue(a.Ctx, gcfg.ContextKeyOperation, operation)
+	return a
+}
+
+// WithNamespace sets the namespace in the context
+func (a *ApolloAdapterCtx) WithNamespace(namespace string) *ApolloAdapterCtx {
+	a.Ctx = context.WithValue(a.Ctx, ContextKeyNamespace, namespace)
+	return a
+}
+
+// WithAppId sets the appId in the context
+func (a *ApolloAdapterCtx) WithAppId(appId string) *ApolloAdapterCtx {
+	a.Ctx = context.WithValue(a.Ctx, ContextKeyAppId, appId)
+	return a
+}
+
+// WithCluster sets the cluster in the context
+func (a *ApolloAdapterCtx) WithCluster(cluster string) *ApolloAdapterCtx {
+	a.Ctx = context.WithValue(a.Ctx, ContextKeyCluster, cluster)
+	return a
+}
+
+// WithContent sets the content in the context
+func (a *ApolloAdapterCtx) WithContent(content *gjson.Json) *ApolloAdapterCtx {
+	a.Ctx = context.WithValue(a.Ctx, gcfg.ContextKeyContent, content)
+	return a
+}
+
+// GetNamespace retrieves the namespace from the context
+func (a *ApolloAdapterCtx) GetNamespace() string {
+	if v := a.Ctx.Value(ContextKeyNamespace); v != nil {
+		if s, ok := v.(string); ok {
+			return s
+		}
+	}
+	return ""
+}
+
+// GetAppId retrieves the appId from the context
+func (a *ApolloAdapterCtx) GetAppId() string {
+	if v := a.Ctx.Value(ContextKeyAppId); v != nil {
+		if s, ok := v.(string); ok {
+			return s
+		}
+	}
+	return ""
+}
+
+// GetCluster retrieves the cluster from the context
+func (a *ApolloAdapterCtx) GetCluster() string {
+	if v := a.Ctx.Value(ContextKeyCluster); v != nil {
+		if s, ok := v.(string); ok {
+			return s
+		}
+	}
+	return ""
+}
+
+// GetContent retrieves the content from the context
+func (a *ApolloAdapterCtx) GetContent() *gjson.Json {
+	if v := a.Ctx.Value(gcfg.ContextKeyContent); v != nil {
+		if s, ok := v.(*gjson.Json); ok {
+			return s
+		}
+	}
+	return gjson.New(nil)
+}
+
+// GetOperation retrieves the operation from the context
+func (a *ApolloAdapterCtx) GetOperation() gcfg.OperationType {
+	if v := a.Ctx.Value(gcfg.ContextKeyOperation); v != nil {
+		if s, ok := v.(gcfg.OperationType); ok {
+			return s
+		}
+	}
+	return ""
+}

contrib/config/consul/consul.go

@@ -21,6 +21,12 @@ import (
 	"github.com/gogf/gf/v2/os/glog"
 )
 
+var (
+	// Compile-time checking for interface implementation.
+	_ gcfg.Adapter        = (*Client)(nil)
+	_ gcfg.WatcherAdapter = (*Client)(nil)
+)
+
 // Config is the configuration object for consul client.
 type Config struct {
 	// api.Config in consul package
@@ -41,6 +47,8 @@ type Client struct {
 	client *api.Client
 	// Configmap content cached. It is `*gjson.Json` value internally.
 	value *g.Var
+	// Watchers for watching file changes.
+	watchers *gcfg.WatcherRegistry
 }
 
 // New creates and returns gcfg.Adapter implementing using consul service.
@@ -55,8 +63,9 @@ func New(ctx context.Context, config Config) (adapter gcfg.Adapter, err error) {
 	}
 
 	client := &Client{
-		config: config,
-		value:  g.NewVar(nil, true),
+		config:   config,
+		value:    g.NewVar(nil, true),
+		watchers: gcfg.NewWatcherRegistry(),
 	}
 
 	client.client, err = api.NewClient(&config.ConsulConfig)
@@ -156,13 +165,26 @@ func (c *Client) addWatcher() (err error) {
 		if v, ok = raw.(*api.KVPair); !ok {
 			return
 		}
-
-		if err = c.doUpdate(v.Value); err != nil {
+		err = c.doUpdate(v.Value)
+		if err != nil {
 			c.config.Logger.Errorf(
 				context.Background(),
 				"watch config from consul path %+v update failed: %s",
 				c.config.Path, err,
 			)
+		} else {
+			var m *gjson.Json
+			m, err = gjson.LoadContent(v.Value, true)
+			if err != nil {
+				c.config.Logger.Errorf(
+					context.Background(),
+					"watch config from consul path %+v parse failed: %s",
+					c.config.Path, err,
+				)
+			} else {
+				adapterCtx := NewAdapterCtx().WithOperation(gcfg.OperationUpdate).WithPath(c.config.Path).WithContent(m)
+				c.notifyWatchers(adapterCtx.Ctx)
+			}
 		}
 	}
 
@@ -173,6 +195,7 @@ func (c *Client) addWatcher() (err error) {
 	return nil
 }
 
+// startAsynchronousWatch starts the asynchronous watch.
 func (c *Client) startAsynchronousWatch(plan *watch.Plan) {
 	if err := plan.Run(c.config.ConsulConfig.Address); err != nil {
 		c.config.Logger.Errorf(
@@ -182,3 +205,23 @@ func (c *Client) startAsynchronousWatch(plan *watch.Plan) {
 		)
 	}
 }
+
+// AddWatcher adds a watcher for the specified configuration file.
+func (c *Client) AddWatcher(name string, f func(ctx context.Context)) {
+	c.watchers.Add(name, f)
+}
+
+// RemoveWatcher removes the watcher for the specified configuration file.
+func (c *Client) RemoveWatcher(name string) {
+	c.watchers.Remove(name)
+}
+
+// GetWatcherNames returns all watcher names.
+func (c *Client) GetWatcherNames() []string {
+	return c.watchers.GetNames()
+}
+
+// notifyWatchers notifies all watchers.
+func (c *Client) notifyWatchers(ctx context.Context) {
+	c.watchers.Notify(ctx)
+}