Merge pull request #149 from brzyangg/master

自动 watchUpdate 热加载配置

cache/cache.go

@@ -231,4 +231,5 @@ func init() {
 	} else {
 		slog.Infof(ctx, "%s cache configer:%v been set", fun, constants.ConfigerTypeApollo)
 	}
+	WatchUpdate(ctx)
 }

cache/cache_test.go

@@ -3,8 +3,10 @@ package cache
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"github.com/stretchr/testify/assert"
 	"testing"
+	"time"
 )
 
 type Test struct {
@@ -62,22 +64,27 @@ func TestCache_Set(t *testing.T) {
 
 func TestCache_Get(t *testing.T) {
 	ctx := context.Background()
-	c, err  := NewCacheByNamespace(ctx, "test/test","test",60)
+	c, err  := NewCacheByNamespace(ctx, "base/test","test",60)
 	if err != nil {
 		t.Errorf("NewCacheByNamespace err: %s", err.Error())
 	}
-	WatchUpdate(ctx)
+	//WatchUpdate(ctx)
 	var test Test
-	err = c.Get("test", &test)
-	if err != nil {
-		t.Errorf("Get err: %s", err.Error())
-	}
-	t.Log(test)
-	c.Del("test")
-	// load
-	err = c.Get("test", &test)
-	if err != nil {
-		t.Errorf("Get err: %s", err.Error())
+	for {
+		err = c.Get("test", &test)
+		if err != nil {
+			t.Errorf("Get err: %s", err.Error())
+			return
+		}
+		fmt.Println(test)
+		time.Sleep(time.Second)
 	}
-	assert.Equal(t, 1, test.ID)
+	//t.Log(test)
+	//c.Del("test")
+	//// load
+	//err = c.Get("test", &test)
+	//if err != nil {
+	//	t.Errorf("Get err: %s", err.Error())
+	//}
+	//assert.Equal(t, 1, test.ID)
 }

cache/redis/instance.go

@@ -159,6 +159,11 @@ func (m *InstanceManager) applyChangeEvent(ctx context.Context, ce *center.Chang
 
 func (m *InstanceManager) Watch(ctx context.Context) {
 	fun := "InstanceManager.Watch-->"
+	defer func() {
+		if r := recover(); r != nil {
+			slog.Errorf(ctx, "%s recover r: %v", fun, r)
+		}
+	}()
 	m.watchOnce.Do(func() {
 		slog.Infof(ctx, "%s start watching updates", fun)
 		ceChan := DefaultConfiger.Watch(ctx)

cache/redisext/redisext.go

@@ -3,6 +3,7 @@ package redisext
 import (
 	"context"
 	"fmt"
+	"github.com/shawnfeng/sutil/cache"
 	"time"
 
 	redis2 "github.com/go-redis/redis"
@@ -917,4 +918,5 @@ func init() {
 	} else {
 		slog.Infof(ctx, "%s redisext configer:%v been set", fun, constants.ConfigerTypeApollo)
 	}
+	WatchUpdate(ctx)
 }

cache/value/value.go

@@ -276,4 +276,5 @@ func init() {
 	} else {
 		slog.Infof(ctx, "%s cache configer:%v been set", fun, constants.ConfigerTypeApollo)
 	}
+	WatchUpdate(ctx)
 }

mq/examples/main.go

@@ -23,7 +23,7 @@ type Msg struct {
 func main() {
 
 	_ = trace.InitDefaultTracer("mq.test")
-	topic := "palfish.test.test1"
+	topic := "palfish.test.test"
 
 	ctx := context.Background()
 	ctx = context.WithValue(ctx, scontext.ContextKeyHead, "hahahaha")
@@ -33,8 +33,8 @@ func main() {
 		defer span.Finish()
 	}
 
-	_ = mq.SetConfiger(ctx, mq.ConfigerTypeApollo)
-	mq.WatchUpdate(ctx)
+	//_ = mq.SetConfiger(ctx, mq.ConfigerTypeApollo)
+	//mq.WatchUpdate(ctx)
 
 	/*
 		go func() {

mq/instance.go

@@ -231,6 +231,11 @@ func (m *InstanceManager) applyChangeEvent(ctx context.Context, ce *center.Chang
 
 func (m *InstanceManager) watch(ctx context.Context) {
 	fun := "InstanceManager.watch-->"
+	defer func() {
+		if r := recover(); r != nil {
+			slog.Errorf(ctx, "%s recover r: %v", fun, r)
+		}
+	}()
 	m.watchOnce.Do(func() {
 		slog.Infof(ctx, "%s start watching updates", fun)
 		ceChan := DefaultConfiger.Watch(ctx)

mq/interface.go

@@ -456,4 +456,5 @@ func init() {
 	} else {
 		slog.Infof(ctx, "%s mq configer:%v been set", fun, ConfigerTypeApollo)
 	}
+	WatchUpdate(ctx)
 }