This is an automated email from the ASF dual-hosted git repository.
hsluoyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/casbin-redis-watcher.git
The following commit(s) were added to refs/heads/master by this push:
new b72099c fix: stop watcher without publishing a shutdown message (#62)
b72099c is described below
commit b72099cf822b61c6f82c4d33fc3880b7f8402a42
Author: CodingCaius <[email protected]>
AuthorDate: Mon Oct 19 22:43:48 2026 +0800
fix: stop watcher without publishing a shutdown message (#62)
---
watcher.go | 16 ++++++--
watcher_test.go | 115 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++
2 files changed, 128 insertions(+), 3 deletions(-)
diff --git a/watcher.go b/watcher.go
index ecb4854..2cc68e3 100644
--- a/watcher.go
+++ b/watcher.go
@@ -21,6 +21,7 @@ type Watcher struct {
pubClient rds.UniversalClient
options WatcherOptions
close chan struct{}
+ closeOnce sync.Once
callback func(string)
ctx context.Context
}
@@ -445,7 +446,17 @@ func (w *Watcher) subscribe() {
}()
ch := sub.Channel()
wg.Done()
- for msg := range ch {
+ for {
+ var msg *rds.Message
+ select {
+ case <-w.close:
+ return
+ case message, ok := <-ch:
+ if !ok {
+ return
+ }
+ msg = message
+ }
select {
case <-w.close:
return
@@ -476,6 +487,5 @@ func (w *Watcher) GetWatcherOptions() WatcherOptions {
func (w *Watcher) Close() {
w.l.Lock()
defer w.l.Unlock()
- close(w.close)
- w.pubClient.Publish(w.ctx, w.options.Channel, "Close")
+ w.closeOnce.Do(func() { close(w.close) })
}
diff --git a/watcher_test.go b/watcher_test.go
index 701ab5f..2f3e531 100644
--- a/watcher_test.go
+++ b/watcher_test.go
@@ -1,9 +1,12 @@
package rediswatcher_test
import (
+ "context"
"encoding/json"
+ "errors"
"fmt"
"reflect"
+ "sync"
"testing"
"time"
@@ -14,6 +17,118 @@ import (
"github.com/redis/go-redis/v9"
)
+func TestWatcherCloseDoesNotPublish(t *testing.T) {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ client := redis.NewClient(&redis.Options{Addr: "127.0.0.1:6379"})
+ defer client.Close()
+ channel := "/casbin-close-" + uuid.NewString()
+ observer := client.Subscribe(ctx, channel)
+ defer observer.Close()
+ if _, err := observer.Receive(ctx); err != nil {
+ t.Fatal(err)
+ }
+ messages := observer.Channel()
+ pubClient := redis.NewClient(&redis.Options{Addr: "127.0.0.1:6379"})
+ subClient := redis.NewClient(&redis.Options{Addr: "127.0.0.1:6379"})
+ watcher, err := rediswatcher.NewWatcher("", rediswatcher.WatcherOptions{
+ Channel: channel, PubClient: pubClient, SubClient: subClient,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ closed := false
+ defer func() {
+ if !closed {
+ watcher.Close()
+ }
+ }()
+ updates := make(chan string, 1)
+ survivorID := uuid.NewString()
+ survivor, err := rediswatcher.NewWatcher("127.0.0.1:6379",
rediswatcher.WatcherOptions{
+ Channel: channel,
+ LocalID: survivorID,
+ OptionalUpdateCallback: func(message string) {
+ updates <- message
+ },
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer survivor.Close()
+ waitForWatcherCondition(t, ctx, func() bool {
+ counts, err := client.PubSubNumSub(ctx, channel).Result()
+ return err == nil && counts[channel] == 3
+ })
+ watcher.Close()
+ closed = true
+ waitForWatcherCondition(t, ctx, func() bool {
+ counts, err := client.PubSubNumSub(ctx, channel).Result()
+ return err == nil && counts[channel] == 2 &&
+ errors.Is(pubClient.Ping(ctx).Err(), redis.ErrClosed) &&
+ errors.Is(subClient.Ping(ctx).Err(), redis.ErrClosed)
+ })
+ if err := survivor.Update(); err != nil {
+ t.Fatal(err)
+ }
+ select {
+ case message := <-messages:
+ var update rediswatcher.MSG
+ if err := json.Unmarshal([]byte(message.Payload), &update); err
!= nil {
+ t.Fatalf("Close published a non-JSON message: %q",
message.Payload)
+ }
+ if update.Method != rediswatcher.Update || update.ID !=
survivorID {
+ t.Fatalf("unexpected message after Close: %q",
message.Payload)
+ }
+ case <-ctx.Done():
+ t.Fatal("surviving watcher did not publish an update")
+ }
+ select {
+ case <-updates:
+ case <-ctx.Done():
+ t.Fatal("surviving watcher did not receive an update")
+ }
+}
+
+func TestWatcherCloseIsIdempotent(t *testing.T) {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ pubClient := redis.NewClient(&redis.Options{Addr: "127.0.0.1:6379"})
+ subClient := redis.NewClient(&redis.Options{Addr: "127.0.0.1:6379"})
+ watcher, err := rediswatcher.NewWatcher("", rediswatcher.WatcherOptions{
+ Channel: "/casbin-close-" + uuid.NewString(), PubClient:
pubClient, SubClient: subClient,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ var wg sync.WaitGroup
+ for i := 0; i < 10; i++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ watcher.Close()
+ }()
+ }
+ wg.Wait()
+ waitForWatcherCondition(t, ctx, func() bool {
+ return errors.Is(pubClient.Ping(ctx).Err(), redis.ErrClosed) &&
+ errors.Is(subClient.Ping(ctx).Err(), redis.ErrClosed)
+ })
+}
+
+func waitForWatcherCondition(t *testing.T, ctx context.Context, condition
func() bool) {
+ t.Helper()
+ ticker := time.NewTicker(10 * time.Millisecond)
+ defer ticker.Stop()
+ for !condition() {
+ select {
+ case <-ctx.Done():
+ t.Fatal("timed out waiting for watcher lifecycle
condition")
+ case <-ticker.C:
+ }
+ }
+}
+
func initWatcherWithOptions(t *testing.T, wo rediswatcher.WatcherOptions,
cluster ...bool) (*casbin.Enforcer, *rediswatcher.Watcher) {
var (
w persist.Watcher
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
