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-informer-watcher.git
The following commit(s) were added to refs/heads/master by this push:
new 0e911cf feat: support incremental policy updates from CRD events
0e911cf is described below
commit 0e911cfad7396bc2f285a83b0f5213d967a97f7c
Author: Yang Luo <[email protected]>
AuthorDate: Wed Sep 9 22:39:31 2026 +0800
feat: support incremental policy updates from CRD events
---
README.md | 71 +++++++-
event_test.go | 549 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++
options.go | 7 +
policy.go | 196 ++++++++++++++++++++
policy_test.go | 269 +++++++++++++++++++++++++++
watcher.go | 213 +++++++++++++++++++---
watcher_test.go | 4 +-
7 files changed, 1272 insertions(+), 37 deletions(-)
diff --git a/README.md b/README.md
index eb84f35..20a958a 100644
--- a/README.md
+++ b/README.md
@@ -130,9 +130,10 @@ func main() {
// Create watcher with custom options
options := informerwatcher.WatcherOptions{
- LocalID: "instance-1", // Custom instance
identifier
- IgnoreSelf: true, // Ignore updates from
this instance
- ResyncPeriod: 30 * time.Second, // Resync period with API
server
+ LocalID: "instance-1", // Custom instance
identifier
+ IgnoreSelf: true, // Ignore updates from
this instance
+ ResyncPeriod: 30 * time.Second, // Resync period with API
server
+ IncrementalUpdate: true, // Report the rules that
changed instead of reloading
}
watcher, err := informerwatcher.NewWatcher(client, gvr, "default",
options)
@@ -221,16 +222,68 @@ func main() {
- **LocalID** (string): Unique identifier for this watcher instance.
Auto-generated if not provided.
- **IgnoreSelf** (bool): If true, ignores updates triggered by this watcher
instance. Default: false.
- **ResyncPeriod** (time.Duration): Period for the informer to resync with the
API server. Default: 30 seconds.
-- **OptionalUpdateCallback** (func(string)): Optional callback function set
during initialization.
+- **OptionalUpdateCallback** (func(string)): Optional callback function set
during initialization. Equivalent to calling `SetUpdateCallback` right after
`NewWatcher`, but it is in place before the first event arrives.
+- **IncrementalUpdate** (bool): If true, the watcher reports the exact rules
that changed instead of asking the enforcer to reload its whole policy.
Default: false. See [Update Modes](#update-modes).
## How It Works
1. The watcher uses Kubernetes informers to monitor Custom Resource
Definitions (CRDs) that represent Casbin policies.
-2. When a CRD is created, updated, or deleted, the informer triggers the
corresponding event handler.
-3. The event handler processes the change and invokes the registered callback
function.
-4. The callback typically triggers the enforcer to reload its policy or apply
incremental updates.
+2. When a policy resource is created, updated, or deleted, the informer
triggers the corresponding event handler.
+3. The event handler turns the change into one or more watcher messages and
invokes the registered callback function.
+4. `DefaultUpdateCallback` applies each message to the enforcer, either by
reloading the policy or by applying the incremental change.
5. All running instances with the same watcher configuration receive the same
updates, keeping policies synchronized.
+Three kinds of event carry no change and are dropped rather than reported:
+
+- The resources that already exist when the watcher starts. The informer
replays them as create events while it builds its cache, but they describe the
state the enforcer loaded through its own adapter.
+- The periodic resync, which re-delivers every cached resource. Kubernetes
only bumps `resourceVersion` on a real write, so a resource that comes back
with the same one has not changed.
+- Updates from this instance itself, when `IgnoreSelf` is enabled and the
resource carries the `casbin.org/source-id` annotation with this watcher's
`LocalID`.
+
+Deletions missed while the watch was down arrive as a tombstone rather than
the resource itself; the watcher unwraps it and reports the deletion normally.
+
+## Update Modes
+
+By default, every observed change asks the enforcer for a full reload (the
`Update` message). Reloading is idempotent, so instances cannot drift apart no
matter how a resource was edited. This is the recommended mode when the
enforcer's adapter reads the same resources the watcher observes.
+
+With `IncrementalUpdate` enabled, the watcher reads the rules out of the
resource and reports only what changed:
+
+- A create or delete becomes an add or remove of the rules the resource
carries, batched per section and policy type.
+- An edit is diffed: the rules that disappeared are removed and the rules that
appeared are added. A resource that gains or loses a line is handled the same
way as one whose lines were rewritten.
+- Anything that cannot be described this way — a resource that does not follow
the layout below, or a malformed policy line — falls back to a full reload, so
an unrecognized change is never silently dropped.
+
+Incremental mode requires policy resources to follow the [policy resource
layout](#policy-resource-layout). Note that Casbin applies these messages
through its `Self*` APIs, which write through to the adapter when `AutoSave` is
enabled: if your adapter is backed by the same resources the watcher observes,
keep the default reload mode so the observed change is not written straight
back.
+
+## Policy Resource Layout
+
+In incremental mode the watcher reads rules from two `spec` fields. Both hold
Casbin policy lines in the same format used in a policy CSV file, so the policy
type comes first and the section is its first character (`p` and `p2` belong to
section `p`, `g` and `g2` to section `g`). Blank lines and `#` comments are
skipped.
+
+A single rule, or several newline-separated ones:
+
+```yaml
+apiVersion: casbin.org/v1
+kind: Policy
+metadata:
+ name: alice-can-read
+spec:
+ policy: "p, alice, data1, read"
+```
+
+Or a list:
+
+```yaml
+apiVersion: casbin.org/v1
+kind: Policy
+metadata:
+ name: team-policies
+spec:
+ policies:
+ - "p, alice, data1, read"
+ - "p, bob, data2, write"
+ - "g, alice, admin"
+```
+
+The default reload mode does not read these fields, and works with any
resource layout.
+
## Supported Update Types
The watcher supports all standard Casbin policy update operations:
@@ -269,6 +322,10 @@ spec:
properties:
policy:
type: string
+ policies:
+ type: array
+ items:
+ type: string
scope: Namespaced
names:
plural: policies
diff --git a/event_test.go b/event_test.go
new file mode 100644
index 0000000..2a7771a
--- /dev/null
+++ b/event_test.go
@@ -0,0 +1,549 @@
+// Copyright 2026 The casbin Authors. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package informerwatcher
+
+import (
+ "reflect"
+ "sync"
+ "testing"
+
+ "github.com/casbin/casbin/v2"
+ "github.com/casbin/casbin/v2/model"
+ "github.com/casbin/casbin/v2/persist"
+ "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+ "k8s.io/client-go/tools/cache"
+)
+
+const testModel = `
+[request_definition]
+r = sub, obj, act
+
+[policy_definition]
+p = sub, obj, act
+
+[role_definition]
+g = _, _
+
+[policy_effect]
+e = some(where (p.eft == allow))
+
+[matchers]
+m = g(r.sub, p.sub) && r.obj == p.obj && r.act == p.act
+`
+
+// collector records the messages a watcher hands to its callback.
+type collector struct {
+ mu sync.Mutex
+ msgs []*MSG
+}
+
+func (c *collector) callback(msg string) {
+ decoded := &MSG{}
+ if err := decoded.UnmarshalBinary([]byte(msg)); err != nil {
+ panic(err)
+ }
+
+ c.mu.Lock()
+ defer c.mu.Unlock()
+
+ c.msgs = append(c.msgs, decoded)
+}
+
+func (c *collector) take() []*MSG {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+
+ msgs := c.msgs
+ c.msgs = nil
+
+ return msgs
+}
+
+func newTestWatcher(t *testing.T, options WatcherOptions) (*Watcher,
*collector) {
+ t.Helper()
+
+ msgs := &collector{}
+ options.OptionalUpdateCallback = msgs.callback
+
+ watcher, err := NewWatcher(createTestClient(), testGVR, "", options)
+ if err != nil {
+ t.Fatalf("Failed to create watcher: %v", err)
+ }
+
+ t.Cleanup(watcher.Close)
+
+ return watcher.(*Watcher), msgs
+}
+
+func withVersion(obj *unstructured.Unstructured, version string)
*unstructured.Unstructured {
+ obj.SetResourceVersion(version)
+ return obj
+}
+
+// TestOptionalUpdateCallback tests that a callback passed through the options
+// is used without a separate SetUpdateCallback call.
+func TestOptionalUpdateCallback(t *testing.T) {
+ w, msgs := newTestWatcher(t, WatcherOptions{})
+
+ w.handleEvent(eventAdd, nil, newPolicyResource(map[string]interface{}{
+ "policy": "p, alice, data1, read",
+ }))
+
+ if got := msgs.take(); len(got) != 1 {
+ t.Fatalf("Expected 1 message, got %d", len(got))
+ }
+}
+
+// TestNewWatcherWithNilClient tests that an unusable configuration is
rejected.
+func TestNewWatcherWithNilClient(t *testing.T) {
+ if _, err := NewWatcher(nil, testGVR, "", WatcherOptions{}); err == nil
{
+ t.Error("Expected an error when the dynamic client is nil")
+ }
+}
+
+// TestReloadOnChange tests that, by default, every change asks the enforcer
for
+// a full reload.
+func TestReloadOnChange(t *testing.T) {
+ w, msgs := newTestWatcher(t, WatcherOptions{})
+
+ obj := withVersion(newPolicyResource(map[string]interface{}{
+ "policy": "p, alice, data1, read",
+ }), "1")
+ updated := withVersion(newPolicyResource(map[string]interface{}{
+ "policy": "p, alice, data1, write",
+ }), "2")
+
+ w.handleEvent(eventAdd, nil, obj)
+ w.handleEvent(eventUpdate, obj, updated)
+ w.handleEvent(eventDelete, updated, nil)
+
+ got := msgs.take()
+ if len(got) != 3 {
+ t.Fatalf("Expected 3 messages, got %d", len(got))
+ }
+
+ for i, msg := range got {
+ if msg.Method != Update {
+ t.Errorf("Message %d: expected method %q, got %q", i,
Update, msg.Method)
+ }
+ }
+}
+
+// TestIncrementalAddAndDelete tests that a created or removed resource is
+// translated into the matching policy operations.
+func TestIncrementalAddAndDelete(t *testing.T) {
+ tests := []struct {
+ name string
+ wantSingle UpdateType
+ wantBatch UpdateType
+ }{
+ {"add", UpdateForAddPolicy, UpdateForAddPolicies},
+ {"delete", UpdateForRemovePolicy, UpdateForRemovePolicies},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ w, msgs := newTestWatcher(t,
WatcherOptions{IncrementalUpdate: true})
+
+ single := newPolicyResource(map[string]interface{}{
+ "policy": "p, alice, data1, read",
+ })
+ batch := newPolicyResource(map[string]interface{}{
+ "policies": []interface{}{
+ "p, alice, data1, read",
+ "p, bob, data2, write",
+ "g, alice, admin",
+ },
+ })
+
+ fire := func(obj *unstructured.Unstructured) {
+ if tt.name == "add" {
+ w.handleEvent(eventAdd, nil, obj)
+ return
+ }
+ w.handleEvent(eventDelete, obj, nil)
+ }
+
+ fire(single)
+
+ got := msgs.take()
+ if len(got) != 1 {
+ t.Fatalf("Expected 1 message, got %d", len(got))
+ }
+ if got[0].Method != tt.wantSingle {
+ t.Errorf("Expected method %q, got %q",
tt.wantSingle, got[0].Method)
+ }
+ if !reflect.DeepEqual(got[0].NewRule, []string{"alice",
"data1", "read"}) {
+ t.Errorf("Unexpected rule: %v", got[0].NewRule)
+ }
+
+ fire(batch)
+
+ // The p rules batch together; the g rule is its own
section.
+ got = msgs.take()
+ if len(got) != 2 {
+ t.Fatalf("Expected 2 messages, got %d",
len(got))
+ }
+
+ if got[0].Method != tt.wantBatch || got[0].Sec != "p"
|| got[0].Ptype != "p" {
+ t.Errorf("Unexpected p message: %+v", got[0])
+ }
+ if len(got[0].NewRules) != 2 {
+ t.Errorf("Expected 2 batched p rules, got %v",
got[0].NewRules)
+ }
+
+ if got[1].Method != tt.wantSingle || got[1].Sec != "g"
|| got[1].Ptype != "g" {
+ t.Errorf("Unexpected g message: %+v", got[1])
+ }
+ })
+ }
+}
+
+// TestIncrementalUpdate tests that an edited resource is translated into the
+// rules that were actually removed and added.
+func TestIncrementalUpdate(t *testing.T) {
+ tests := []struct {
+ name string
+ old []interface{}
+ new []interface{}
+ want []*MSG
+ }{
+ {
+ name: "rule replaced",
+ old: []interface{}{"p, alice, data1, read"},
+ new: []interface{}{"p, alice, data1, write"},
+ want: []*MSG{
+ {Method: UpdateForRemovePolicy, Sec: "p",
Ptype: "p", NewRule: []string{"alice", "data1", "read"}},
+ {Method: UpdateForAddPolicy, Sec: "p", Ptype:
"p", NewRule: []string{"alice", "data1", "write"}},
+ },
+ },
+ {
+ name: "rule added",
+ old: []interface{}{"p, alice, data1, read"},
+ new: []interface{}{"p, alice, data1, read", "p, bob,
data2, write"},
+ want: []*MSG{
+ {Method: UpdateForAddPolicy, Sec: "p", Ptype:
"p", NewRule: []string{"bob", "data2", "write"}},
+ },
+ },
+ {
+ name: "rule removed",
+ old: []interface{}{"p, alice, data1, read", "p, bob,
data2, write"},
+ new: []interface{}{"p, alice, data1, read"},
+ want: []*MSG{
+ {Method: UpdateForRemovePolicy, Sec: "p",
Ptype: "p", NewRule: []string{"bob", "data2", "write"}},
+ },
+ },
+ {
+ name: "resource emptied",
+ old: []interface{}{"p, alice, data1, read", "p, bob,
data2, write"},
+ new: []interface{}{},
+ want: []*MSG{
+ {Method: UpdateForRemovePolicies, Sec: "p",
Ptype: "p", NewRules: [][]string{
+ {"alice", "data1", "read"},
+ {"bob", "data2", "write"},
+ }},
+ },
+ },
+ {
+ name: "rules reordered only",
+ old: []interface{}{"p, alice, data1, read", "p, bob,
data2, write"},
+ new: []interface{}{"p, bob, data2, write", "p, alice,
data1, read"},
+ want: nil,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ w, msgs := newTestWatcher(t,
WatcherOptions{IncrementalUpdate: true})
+
+ oldObj :=
withVersion(newPolicyResource(map[string]interface{}{"policies": tt.old}), "1")
+ newObj :=
withVersion(newPolicyResource(map[string]interface{}{"policies": tt.new}), "2")
+
+ w.handleEvent(eventUpdate, oldObj, newObj)
+
+ got := msgs.take()
+ if len(got) != len(tt.want) {
+ t.Fatalf("Expected %d messages, got %d: %+v",
len(tt.want), len(got), got)
+ }
+
+ for i, want := range tt.want {
+ want.ID = w.localID
+
+ if !reflect.DeepEqual(got[i], want) {
+ t.Errorf("Message %d: expected %+v, got
%+v", i, want, got[i])
+ }
+ }
+ })
+ }
+}
+
+// TestResyncIsIgnored tests that the periodic resync, which re-delivers every
+// cached resource unchanged, produces no messages.
+func TestResyncIsIgnored(t *testing.T) {
+ for _, incremental := range []bool{false, true} {
+ w, msgs := newTestWatcher(t, WatcherOptions{IncrementalUpdate:
incremental})
+
+ obj := withVersion(newPolicyResource(map[string]interface{}{
+ "policy": "p, alice, data1, read",
+ }), "42")
+
+ w.handleEvent(eventUpdate, obj, obj.DeepCopy())
+
+ if got := msgs.take(); len(got) != 0 {
+ t.Errorf("IncrementalUpdate=%v: expected no messages
for a resync, got %+v", incremental, got)
+ }
+ }
+}
+
+// TestDeleteTombstone tests the tombstone Kubernetes delivers when a deletion
+// was missed while the watch was down.
+func TestDeleteTombstone(t *testing.T) {
+ w, msgs := newTestWatcher(t, WatcherOptions{IncrementalUpdate: true})
+
+ tombstone := cache.DeletedFinalStateUnknown{
+ Key: "default/test-policy",
+ Obj: newPolicyResource(map[string]interface{}{"policy": "p,
alice, data1, read"}),
+ }
+
+ w.handleEvent(eventDelete, tombstone, nil)
+
+ got := msgs.take()
+ if len(got) != 1 {
+ t.Fatalf("Expected 1 message, got %d", len(got))
+ }
+
+ if got[0].Method != UpdateForRemovePolicy {
+ t.Errorf("Expected method %q, got %q", UpdateForRemovePolicy,
got[0].Method)
+ }
+
+ w.handleEvent(eventDelete, cache.DeletedFinalStateUnknown{Key:
"default/gone"}, nil)
+
+ if got := msgs.take(); len(got) != 0 {
+ t.Errorf("Expected no messages for an empty tombstone, got
%+v", got)
+ }
+}
+
+// TestIncrementalFallsBackToReload tests that a change which cannot be
+// described incrementally still reaches the enforcer as a full reload.
+func TestIncrementalFallsBackToReload(t *testing.T) {
+ tests := []struct {
+ name string
+ spec map[string]interface{}
+ }{
+ {"unknown layout", map[string]interface{}{"unexpected":
"field"}},
+ {"malformed policy line", map[string]interface{}{"policy":
"p"}},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ w, msgs := newTestWatcher(t,
WatcherOptions{IncrementalUpdate: true})
+
+ w.handleEvent(eventAdd, nil, newPolicyResource(tt.spec))
+
+ got := msgs.take()
+ if len(got) != 1 {
+ t.Fatalf("Expected 1 message, got %d", len(got))
+ }
+
+ if got[0].Method != Update {
+ t.Errorf("Expected method %q, got %q", Update,
got[0].Method)
+ }
+ })
+ }
+}
+
+// TestNoEventsAfterClose tests that a closed watcher stops reporting changes.
+func TestNoEventsAfterClose(t *testing.T) {
+ w, msgs := newTestWatcher(t, WatcherOptions{})
+
+ w.Close()
+ w.handleEvent(eventAdd, nil, newPolicyResource(map[string]interface{}{
+ "policy": "p, alice, data1, read",
+ }))
+
+ if got := msgs.take(); len(got) != 0 {
+ t.Errorf("Expected no messages after Close, got %+v", got)
+ }
+}
+
+// memoryAdapter stands in for an adapter backed by the same resources the
+// watcher observes: a test changes its contents to mimic a GitOps apply, then
+// delivers the matching informer event.
+type memoryAdapter struct {
+ mu sync.Mutex
+ lines []string
+}
+
+func (a *memoryAdapter) setLines(lines ...string) {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+
+ a.lines = lines
+}
+
+func (a *memoryAdapter) LoadPolicy(m model.Model) error {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+
+ for _, line := range a.lines {
+ if err := persist.LoadPolicyLine(line, m); err != nil {
+ return err
+ }
+ }
+
+ return nil
+}
+
+func (a *memoryAdapter) SavePolicy(model.Model) error { return
nil }
+func (a *memoryAdapter) AddPolicy(string, string, []string) error { return
nil }
+func (a *memoryAdapter) RemovePolicy(string, string, []string) error { return
nil }
+func (a *memoryAdapter) RemoveFilteredPolicy(string, string, int, ...string)
error {
+ return nil
+}
+
+// TestEnforcementFollowsReload tests that create, update and delete events
+// reach enforcement decisions in the default reload mode.
+func TestEnforcementFollowsReload(t *testing.T) {
+ adapter := &memoryAdapter{lines: []string{"p, alice, data1, read"}}
+
+ m, err := model.NewModelFromString(testModel)
+ if err != nil {
+ t.Fatalf("Failed to build model: %v", err)
+ }
+
+ e, err := casbin.NewEnforcer(m, adapter)
+ if err != nil {
+ t.Fatalf("Failed to create enforcer: %v", err)
+ }
+
+ w, _ := newTestWatcher(t, WatcherOptions{})
+ if err := w.SetUpdateCallback(DefaultUpdateCallback(e)); err != nil {
+ t.Fatalf("Failed to set callback: %v", err)
+ }
+
+ assertEnforce(t, e, true, "alice", "data1", "read")
+
+ adapter.setLines("p, alice, data1, write")
+ w.handleEvent(eventUpdate,
+ withVersion(newPolicyResource(map[string]interface{}{"policy":
"p, alice, data1, read"}), "1"),
+ withVersion(newPolicyResource(map[string]interface{}{"policy":
"p, alice, data1, write"}), "2"))
+
+ assertEnforce(t, e, false, "alice", "data1", "read")
+ assertEnforce(t, e, true, "alice", "data1", "write")
+
+ adapter.setLines()
+ w.handleEvent(eventDelete,
+ withVersion(newPolicyResource(map[string]interface{}{"policy":
"p, alice, data1, write"}), "2"), nil)
+
+ assertEnforce(t, e, false, "alice", "data1", "write")
+}
+
+// TestEnforcementFollowsIncrementalUpdates tests that create, update and
delete
+// events reach enforcement decisions in incremental mode, including the role
+// links a grouping rule brings with it.
+func TestEnforcementFollowsIncrementalUpdates(t *testing.T) {
+ m, err := model.NewModelFromString(testModel)
+ if err != nil {
+ t.Fatalf("Failed to build model: %v", err)
+ }
+
+ e, err := casbin.NewEnforcer(m)
+ if err != nil {
+ t.Fatalf("Failed to create enforcer: %v", err)
+ }
+
+ w, _ := newTestWatcher(t, WatcherOptions{IncrementalUpdate: true})
+ if err := w.SetUpdateCallback(DefaultUpdateCallback(e)); err != nil {
+ t.Fatalf("Failed to set callback: %v", err)
+ }
+
+ direct := map[string]interface{}{"policy": "p, alice, data1, read"}
+ w.handleEvent(eventAdd, nil, withVersion(newPolicyResource(direct),
"1"))
+ assertEnforce(t, e, true, "alice", "data1", "read")
+
+ // Access granted indirectly, so the grouping rule has to rebuild the
+ // enforcer's role links.
+ roles := map[string]interface{}{
+ "policies": []interface{}{"g, alice, admin", "p, admin, data2,
write"},
+ }
+ w.handleEvent(eventAdd, nil, withVersion(newPolicyResource(roles), "1"))
+ assertEnforce(t, e, true, "alice", "data2", "write")
+
+ swapped := map[string]interface{}{"policy": "p, alice, data1, write"}
+ w.handleEvent(eventUpdate,
+ withVersion(newPolicyResource(direct), "1"),
+ withVersion(newPolicyResource(swapped), "2"))
+
+ assertEnforce(t, e, false, "alice", "data1", "read")
+ assertEnforce(t, e, true, "alice", "data1", "write")
+
+ w.handleEvent(eventDelete, withVersion(newPolicyResource(roles), "1"),
nil)
+ assertEnforce(t, e, false, "alice", "data2", "write")
+ assertEnforce(t, e, true, "alice", "data1", "write")
+}
+
+// TestConcurrentEventsWithSyncedEnforcer tests that concurrent events stay
safe
+// when the callback drives a SyncedEnforcer. Run with -race.
+func TestConcurrentEventsWithSyncedEnforcer(t *testing.T) {
+ m, err := model.NewModelFromString(testModel)
+ if err != nil {
+ t.Fatalf("Failed to build model: %v", err)
+ }
+
+ e, err := casbin.NewSyncedEnforcer(m)
+ if err != nil {
+ t.Fatalf("Failed to create enforcer: %v", err)
+ }
+
+ w, _ := newTestWatcher(t, WatcherOptions{IncrementalUpdate: true})
+ if err := w.SetUpdateCallback(DefaultUpdateCallback(e)); err != nil {
+ t.Fatalf("Failed to set callback: %v", err)
+ }
+
+ const users = 20
+
+ var wg sync.WaitGroup
+ for i := 0; i < users; i++ {
+ wg.Add(1)
+
+ go func(i int) {
+ defer wg.Done()
+
+ obj := newPolicyResource(map[string]interface{}{
+ "policy": "p, user" + string(rune('a'+i%26)) +
", data1, read",
+ })
+
+ w.handleEvent(eventAdd, nil, withVersion(obj, "1"))
+ _, _ = e.Enforce("alice", "data1", "read")
+ }(i)
+ }
+
+ wg.Wait()
+
+ assertEnforce(t, e, true, "usera", "data1", "read")
+}
+
+func assertEnforce(t *testing.T, e casbin.IEnforcer, want bool, request
...interface{}) {
+ t.Helper()
+
+ got, err := e.Enforce(request...)
+ if err != nil {
+ t.Fatalf("Enforce%v failed: %v", request, err)
+ }
+
+ if got != want {
+ t.Errorf("Enforce%v: expected %v, got %v", request, want, got)
+ }
+}
diff --git a/options.go b/options.go
index 0bc3637..2ed32bf 100644
--- a/options.go
+++ b/options.go
@@ -33,6 +33,13 @@ type WatcherOptions struct {
// Default is 30 seconds.
ResyncPeriod time.Duration
+ // IncrementalUpdate makes the watcher report the exact rules that
changed
+ // instead of asking the enforcer to reload its whole policy. Leave it
off
+ // when the enforcer's adapter is backed by the same resources the
watcher
+ // observes: Casbin applies these messages through its Self* APIs, which
+ // write through to the adapter when AutoSave is enabled.
+ IncrementalUpdate bool
+
// OptionalUpdateCallback is an optional callback function that can be
set during initialization.
OptionalUpdateCallback func(string)
}
diff --git a/policy.go b/policy.go
new file mode 100644
index 0000000..6826b6b
--- /dev/null
+++ b/policy.go
@@ -0,0 +1,196 @@
+// Copyright 2026 The casbin Authors. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package informerwatcher
+
+import (
+ "encoding/csv"
+ "errors"
+ "fmt"
+ "strings"
+
+ "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+)
+
+// fieldSeparator joins rule fields into a comparable key. It cannot appear in
a
+// policy line, so two rules share a key only if they are the same rule.
+const fieldSeparator = "\x00"
+
+// rule is a single Casbin policy rule carried by a policy resource.
+type rule struct {
+ sec string
+ ptype string
+ fields []string
+}
+
+func (r rule) key() string {
+ return strings.Join(append([]string{r.ptype}, r.fields...),
fieldSeparator)
+}
+
+// extractRules reads the rules held by spec.policy and spec.policies. An error
+// means the resource does not follow that layout, and the caller falls back to
+// a full policy reload.
+func extractRules(obj *unstructured.Unstructured) ([]rule, error) {
+ if obj == nil {
+ return nil, errors.New("policy resource is nil")
+ }
+
+ spec, ok := obj.Object["spec"].(map[string]interface{})
+ if !ok {
+ return nil, errors.New("policy resource has no spec object")
+ }
+
+ lines, err := policyLines(spec)
+ if err != nil {
+ return nil, err
+ }
+
+ rules := make([]rule, 0, len(lines))
+ for _, line := range lines {
+ r, ok, err := parsePolicyLine(line)
+ if err != nil {
+ return nil, err
+ }
+ if !ok {
+ continue
+ }
+
+ rules = append(rules, r)
+ }
+
+ return rules, nil
+}
+
+// policyLines collects the raw policy lines held by spec.policy and
+// spec.policies. At least one of the two must be present.
+func policyLines(spec map[string]interface{}) ([]string, error) {
+ var lines []string
+ found := false
+
+ if raw, ok := spec["policy"]; ok {
+ text, ok := raw.(string)
+ if !ok {
+ return nil, fmt.Errorf("spec.policy must be a string,
got %T", raw)
+ }
+
+ lines = append(lines, strings.Split(text, "\n")...)
+ found = true
+ }
+
+ if raw, ok := spec["policies"]; ok {
+ items, ok := raw.([]interface{})
+ if !ok {
+ return nil, fmt.Errorf("spec.policies must be a list,
got %T", raw)
+ }
+
+ for _, item := range items {
+ text, ok := item.(string)
+ if !ok {
+ return nil, fmt.Errorf("spec.policies must hold
strings, got %T", item)
+ }
+
+ lines = append(lines, text)
+ }
+
+ found = true
+ }
+
+ if !found {
+ return nil, errors.New("policy resource has neither spec.policy
nor spec.policies")
+ }
+
+ return lines, nil
+}
+
+// parsePolicyLine parses one line such as "p, alice, data1, read". It mirrors
+// persist.LoadPolicyLine so a resource and a policy CSV file are read the same
+// way. A blank line or a comment yields ok == false and no error.
+func parsePolicyLine(line string) (rule, bool, error) {
+ line = strings.TrimSpace(line)
+ if line == "" || strings.HasPrefix(line, "#") {
+ return rule{}, false, nil
+ }
+
+ reader := csv.NewReader(strings.NewReader(line))
+ reader.Comma = ','
+ reader.Comment = '#'
+ reader.TrimLeadingSpace = true
+
+ tokens, err := reader.Read()
+ if err != nil {
+ return rule{}, false, fmt.Errorf("parse policy line %q: %w",
line, err)
+ }
+
+ for i := range tokens {
+ tokens[i] = strings.TrimSpace(tokens[i])
+ }
+
+ ptype := tokens[0]
+ if ptype == "" || len(tokens) < 2 {
+ return rule{}, false, fmt.Errorf("policy line %q carries no
rule fields", line)
+ }
+
+ // Casbin takes the section from the first character of the policy
type, so
+ // "p2" belongs to section "p" and "g2" to section "g".
+ return rule{sec: ptype[:1], ptype: ptype, fields: tokens[1:]}, true, nil
+}
+
+// groupRules collects rules by section and policy type, preserving the order
in
+// which each group first appears.
+func groupRules(rules []rule) [][]rule {
+ groups := make([][]rule, 0, len(rules))
+ index := make(map[string]int, len(rules))
+
+ for _, r := range rules {
+ key := r.sec + fieldSeparator + r.ptype
+
+ if i, ok := index[key]; ok {
+ groups[i] = append(groups[i], r)
+ continue
+ }
+
+ index[key] = len(groups)
+ groups = append(groups, []rule{r})
+ }
+
+ return groups
+}
+
+// diffRules returns the rules present in want but missing from have.
+func diffRules(have, want []rule) []rule {
+ seen := make(map[string]struct{}, len(have))
+ for _, r := range have {
+ seen[r.key()] = struct{}{}
+ }
+
+ var missing []rule
+ for _, r := range want {
+ if _, ok := seen[r.key()]; ok {
+ continue
+ }
+
+ missing = append(missing, r)
+ }
+
+ return missing
+}
+
+func ruleFields(rules []rule) [][]string {
+ fields := make([][]string, 0, len(rules))
+ for _, r := range rules {
+ fields = append(fields, r.fields)
+ }
+
+ return fields
+}
diff --git a/policy_test.go b/policy_test.go
new file mode 100644
index 0000000..aae1d0b
--- /dev/null
+++ b/policy_test.go
@@ -0,0 +1,269 @@
+// Copyright 2026 The casbin Authors. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package informerwatcher
+
+import (
+ "reflect"
+ "testing"
+
+ "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+)
+
+// newPolicyResource builds a policy resource carrying the given spec.
+func newPolicyResource(spec map[string]interface{}) *unstructured.Unstructured
{
+ return &unstructured.Unstructured{
+ Object: map[string]interface{}{
+ "apiVersion": "casbin.org/v1",
+ "kind": "Policy",
+ "metadata": map[string]interface{}{
+ "name": "test-policy",
+ "namespace": "default",
+ },
+ "spec": spec,
+ },
+ }
+}
+
+// TestParsePolicyLine tests parsing of individual Casbin policy lines.
+func TestParsePolicyLine(t *testing.T) {
+ tests := []struct {
+ name string
+ line string
+ want rule
+ wantOK bool
+ wantErr bool
+ }{
+ {
+ name: "policy rule",
+ line: "p, alice, data1, read",
+ want: rule{sec: "p", ptype: "p", fields:
[]string{"alice", "data1", "read"}},
+ wantOK: true,
+ },
+ {
+ name: "grouping rule",
+ line: "g, alice, admin",
+ want: rule{sec: "g", ptype: "g", fields:
[]string{"alice", "admin"}},
+ wantOK: true,
+ },
+ {
+ // The section is the first character of the policy
type.
+ name: "numbered grouping rule",
+ line: "g2, alice, admin, domain1",
+ want: rule{sec: "g", ptype: "g2", fields:
[]string{"alice", "admin", "domain1"}},
+ wantOK: true,
+ },
+ {
+ name: "numbered policy rule",
+ line: "p2, alice, data1, read",
+ want: rule{sec: "p", ptype: "p2", fields:
[]string{"alice", "data1", "read"}},
+ wantOK: true,
+ },
+ {
+ name: "quoted field holding a comma",
+ line: `p, alice, "data1,data2", read`,
+ want: rule{sec: "p", ptype: "p", fields:
[]string{"alice", "data1,data2", "read"}},
+ wantOK: true,
+ },
+ {
+ name: "untrimmed line",
+ line: " p , alice , data1 , read ",
+ want: rule{sec: "p", ptype: "p", fields:
[]string{"alice", "data1", "read"}},
+ wantOK: true,
+ },
+ {
+ name: "blank line",
+ line: " ",
+ wantOK: false,
+ },
+ {
+ name: "comment",
+ line: "# p, alice, data1, read",
+ wantOK: false,
+ },
+ {
+ name: "policy type without fields",
+ line: "p",
+ wantErr: true,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ got, ok, err := parsePolicyLine(tt.line)
+
+ if tt.wantErr {
+ if err == nil {
+ t.Fatalf("Expected an error for %q",
tt.line)
+ }
+ return
+ }
+
+ if err != nil {
+ t.Fatalf("Unexpected error for %q: %v",
tt.line, err)
+ }
+
+ if ok != tt.wantOK {
+ t.Fatalf("Expected ok %v, got %v", tt.wantOK,
ok)
+ }
+
+ if ok && !reflect.DeepEqual(got, tt.want) {
+ t.Errorf("Expected rule %+v, got %+v", tt.want,
got)
+ }
+ })
+ }
+}
+
+// TestExtractRules tests reading rules out of a policy resource.
+func TestExtractRules(t *testing.T) {
+ tests := []struct {
+ name string
+ spec map[string]interface{}
+ want []rule
+ wantErr bool
+ }{
+ {
+ name: "single policy",
+ spec: map[string]interface{}{"policy": "p, alice,
data1, read"},
+ want: []rule{{sec: "p", ptype: "p", fields:
[]string{"alice", "data1", "read"}}},
+ },
+ {
+ name: "multi-line policy",
+ spec: map[string]interface{}{"policy": "p, alice,
data1, read\ng, alice, admin\n"},
+ want: []rule{
+ {sec: "p", ptype: "p", fields:
[]string{"alice", "data1", "read"}},
+ {sec: "g", ptype: "g", fields:
[]string{"alice", "admin"}},
+ },
+ },
+ {
+ name: "policy list",
+ spec: map[string]interface{}{
+ "policies": []interface{}{"p, alice, data1,
read", "p, bob, data2, write"},
+ },
+ want: []rule{
+ {sec: "p", ptype: "p", fields:
[]string{"alice", "data1", "read"}},
+ {sec: "p", ptype: "p", fields: []string{"bob",
"data2", "write"}},
+ },
+ },
+ {
+ name: "empty policy list",
+ spec: map[string]interface{}{"policies":
[]interface{}{}},
+ want: []rule{},
+ },
+ {
+ name: "unknown layout",
+ spec: map[string]interface{}{"unexpected": "field"},
+ wantErr: true,
+ },
+ {
+ name: "policy of the wrong type",
+ spec: map[string]interface{}{"policy": 42},
+ wantErr: true,
+ },
+ {
+ name: "policy list holding a non-string",
+ spec: map[string]interface{}{"policies":
[]interface{}{42}},
+ wantErr: true,
+ },
+ {
+ name: "malformed policy line",
+ spec: map[string]interface{}{"policy": "p"},
+ wantErr: true,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ got, err := extractRules(newPolicyResource(tt.spec))
+
+ if tt.wantErr {
+ if err == nil {
+ t.Fatalf("Expected an error, got rules
%+v", got)
+ }
+ return
+ }
+
+ if err != nil {
+ t.Fatalf("Unexpected error: %v", err)
+ }
+
+ if !reflect.DeepEqual(got, tt.want) {
+ t.Errorf("Expected rules %+v, got %+v",
tt.want, got)
+ }
+ })
+ }
+}
+
+// TestExtractRulesWithoutSpec tests a resource that carries no spec at all.
+func TestExtractRulesWithoutSpec(t *testing.T) {
+ obj := &unstructured.Unstructured{Object:
map[string]interface{}{"kind": "Policy"}}
+
+ if _, err := extractRules(obj); err == nil {
+ t.Error("Expected an error for a resource without a spec")
+ }
+
+ if _, err := extractRules(nil); err == nil {
+ t.Error("Expected an error for a nil resource")
+ }
+}
+
+// TestGroupRules tests that rules are grouped by section and policy type.
+func TestGroupRules(t *testing.T) {
+ rules := []rule{
+ {sec: "p", ptype: "p", fields: []string{"alice", "data1",
"read"}},
+ {sec: "g", ptype: "g", fields: []string{"alice", "admin"}},
+ {sec: "p", ptype: "p", fields: []string{"bob", "data2",
"write"}},
+ {sec: "g", ptype: "g2", fields: []string{"admin", "root"}},
+ }
+
+ groups := groupRules(rules)
+
+ if len(groups) != 3 {
+ t.Fatalf("Expected 3 groups, got %d", len(groups))
+ }
+
+ if len(groups[0]) != 2 || groups[0][0].ptype != "p" {
+ t.Errorf("Expected the two p rules in the first group, got
%+v", groups[0])
+ }
+
+ if len(groups[1]) != 1 || groups[1][0].ptype != "g" {
+ t.Errorf("Expected the g rule in the second group, got %+v",
groups[1])
+ }
+
+ if len(groups[2]) != 1 || groups[2][0].ptype != "g2" {
+ t.Errorf("Expected the g2 rule in the third group, got %+v",
groups[2])
+ }
+}
+
+// TestDiffRules tests the rule diff used to translate update events.
+func TestDiffRules(t *testing.T) {
+ alice := rule{sec: "p", ptype: "p", fields: []string{"alice", "data1",
"read"}}
+ bob := rule{sec: "p", ptype: "p", fields: []string{"bob", "data2",
"write"}}
+ // Same fields as alice but a different policy type.
+ alice2 := rule{sec: "p", ptype: "p2", fields: []string{"alice",
"data1", "read"}}
+
+ missing := diffRules([]rule{alice}, []rule{alice, bob, alice2})
+
+ if len(missing) != 2 {
+ t.Fatalf("Expected 2 missing rules, got %d: %+v", len(missing),
missing)
+ }
+
+ if !reflect.DeepEqual(missing[0], bob) ||
!reflect.DeepEqual(missing[1], alice2) {
+ t.Errorf("Unexpected missing rules: %+v", missing)
+ }
+
+ if got := diffRules([]rule{alice, bob}, []rule{alice, bob}); len(got)
!= 0 {
+ t.Errorf("Expected no missing rules, got %+v", got)
+ }
+}
diff --git a/watcher.go b/watcher.go
index 0f94eb4..0984f37 100644
--- a/watcher.go
+++ b/watcher.go
@@ -18,6 +18,7 @@ import (
"context"
"encoding/json"
"errors"
+ "fmt"
"log"
"runtime"
"sync"
@@ -42,8 +43,24 @@ type Watcher struct {
stopCh chan struct{}
ctx context.Context
cancel context.CancelFunc
+
+ // callbackMu serializes callback invocations so the enforcer sees
changes
+ // in the order the informer reported them.
+ callbackMu sync.Mutex
}
+// sourceIDAnnotation marks the watcher instance that wrote a policy resource.
+const sourceIDAnnotation = "casbin.org/source-id"
+
+// eventKind identifies the informer event being translated.
+type eventKind int
+
+const (
+ eventAdd eventKind = iota
+ eventUpdate
+ eventDelete
+)
+
// UpdateType represents the type of policy update.
type UpdateType string
@@ -97,7 +114,7 @@ func DefaultUpdateCallback(e casbin.IEnforcer) func(string) {
switch msgStruct.Method {
case Update, UpdateForSavePolicy:
err = e.LoadPolicy()
- res = true
+ res = err == nil
case UpdateForAddPolicy:
res, err = e.SelfAddPolicy(msgStruct.Sec,
msgStruct.Ptype, msgStruct.NewRule)
case UpdateForAddPolicies:
@@ -118,9 +135,10 @@ func DefaultUpdateCallback(e casbin.IEnforcer)
func(string) {
if err != nil {
log.Println("Error updating policy:", err)
+ return
}
if !res {
- log.Println("Callback update policy failed")
+ log.Println("Callback update policy had no effect")
}
}
}
@@ -132,16 +150,21 @@ func finalizer(w *Watcher) {
// NewWatcher creates a new Watcher instance.
func NewWatcher(client dynamic.Interface, gvr schema.GroupVersionResource,
namespace string, options WatcherOptions) (persist.Watcher, error) {
+ if client == nil {
+ return nil, errors.New("dynamic client cannot be nil")
+ }
+
initConfig(&options)
ctx, cancel := context.WithCancel(context.Background())
w := &Watcher{
- running: true,
- localID: options.LocalID,
- options: options,
- stopCh: make(chan struct{}),
- ctx: ctx,
- cancel: cancel,
+ running: true,
+ localID: options.LocalID,
+ options: options,
+ callback: options.OptionalUpdateCallback,
+ stopCh: make(chan struct{}),
+ ctx: ctx,
+ cancel: cancel,
}
// Create dynamic informer factory
@@ -158,13 +181,21 @@ func NewWatcher(client dynamic.Interface, gvr
schema.GroupVersionResource, names
// Add event handlers
_, err := w.informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
- w.handleEvent(obj, "add")
+ // Resources that already exist are replayed as adds
while the
+ // cache is built; they are state the enforcer already
loaded, not
+ // a change. HasSynced turns true just before the last
replay, so a
+ // stray one may slip through - cheaper than gating any
later and
+ // dropping a resource created moments after startup.
+ if !w.informer.HasSynced() {
+ return
+ }
+ w.handleEvent(eventAdd, nil, obj)
},
UpdateFunc: func(oldObj, newObj interface{}) {
- w.handleEvent(newObj, "update")
+ w.handleEvent(eventUpdate, oldObj, newObj)
},
DeleteFunc: func(obj interface{}) {
- w.handleEvent(obj, "delete")
+ w.handleEvent(eventDelete, obj, nil)
},
})
if err != nil {
@@ -187,45 +218,171 @@ func NewWatcher(client dynamic.Interface, gvr
schema.GroupVersionResource, names
return w, nil
}
-// handleEvent processes CRD events and triggers the callback.
-func (w *Watcher) handleEvent(obj interface{}, eventType string) {
+// handleEvent processes CRD events and triggers the callback. oldObj holds the
+// resource before the event and newObj after it, so an add has no oldObj and a
+// delete has no newObj.
+func (w *Watcher) handleEvent(kind eventKind, oldObj, newObj interface{}) {
w.lock.RLock()
+ running := w.running
callback := w.callback
w.lock.RUnlock()
- if callback == nil {
+ if !running || callback == nil {
return
}
- unstructuredObj, ok := obj.(*unstructured.Unstructured)
- if !ok {
- log.Println("Failed to convert object to unstructured")
+ old, err := asPolicyResource(oldObj)
+ if err != nil {
+ log.Println("Skipping event:", err)
return
}
- // Extract policy update information from the CRD
- msg := &MSG{
- Method: Update,
- ID: w.localID,
+ current, err := asPolicyResource(newObj)
+ if err != nil {
+ log.Println("Skipping event:", err)
+ return
+ }
+
+ // A delete carries only the previous state of the resource.
+ subject := current
+ if subject == nil {
+ subject = old
+ }
+ if subject == nil {
+ return
}
// Check if this is our own update (if IgnoreSelf is enabled)
- if w.options.IgnoreSelf {
- annotations := unstructuredObj.GetAnnotations()
- if annotations != nil {
- if sourceID, ok := annotations["casbin.org/source-id"];
ok && sourceID == w.localID {
- return
- }
+ if w.options.IgnoreSelf && subject.GetAnnotations()[sourceIDAnnotation]
== w.localID {
+ return
+ }
+
+ // A resync re-delivers every cached resource as an update, and
Kubernetes
+ // only bumps resourceVersion on a real write.
+ if kind == eventUpdate && old != nil && current != nil &&
+ old.GetResourceVersion() == current.GetResourceVersion() {
+ return
+ }
+
+ for _, msg := range w.buildMessages(kind, old, current) {
+ w.notify(callback, msg)
+ }
+}
+
+// asPolicyResource converts an object delivered by the informer, unwrapping
the
+// tombstone Kubernetes sends when a deletion was missed while the watch was
+// down. A nil object stays nil: that side of the event does not exist.
+func asPolicyResource(obj interface{}) (*unstructured.Unstructured, error) {
+ if obj == nil {
+ return nil, nil
+ }
+
+ if tombstone, ok := obj.(cache.DeletedFinalStateUnknown); ok {
+ obj = tombstone.Obj
+ }
+
+ resource, ok := obj.(*unstructured.Unstructured)
+ if !ok {
+ return nil, fmt.Errorf("unexpected object type %T in informer
event", obj)
+ }
+
+ return resource, nil
+}
+
+// buildMessages translates an observed change into the messages that describe
+// it: a full reload unless IncrementalUpdate is enabled, otherwise the rules
+// that actually changed.
+func (w *Watcher) buildMessages(kind eventKind, oldObj, newObj
*unstructured.Unstructured) []*MSG {
+ if !w.options.IncrementalUpdate {
+ return []*MSG{w.newMessage(Update)}
+ }
+
+ switch kind {
+ case eventAdd:
+ rules, err := extractRules(newObj)
+ if err != nil {
+ return w.reload(err)
}
+
+ return w.ruleMessages(rules, UpdateForAddPolicy,
UpdateForAddPolicies)
+
+ case eventDelete:
+ rules, err := extractRules(oldObj)
+ if err != nil {
+ return w.reload(err)
+ }
+
+ return w.ruleMessages(rules, UpdateForRemovePolicy,
UpdateForRemovePolicies)
+
+ case eventUpdate:
+ oldRules, err := extractRules(oldObj)
+ if err != nil {
+ return w.reload(err)
+ }
+
+ newRules, err := extractRules(newObj)
+ if err != nil {
+ return w.reload(err)
+ }
+
+ // Diffed rather than paired up: SelfUpdatePolicies needs both
sides to
+ // hold the same number of rules, which a resource that gained
or lost
+ // a line does not.
+ removed := w.ruleMessages(diffRules(newRules, oldRules),
UpdateForRemovePolicy, UpdateForRemovePolicies)
+ added := w.ruleMessages(diffRules(oldRules, newRules),
UpdateForAddPolicy, UpdateForAddPolicies)
+
+ return append(removed, added...)
+
+ default:
+ return nil
}
+}
+
+// reload reports why the change could not be described incrementally.
+func (w *Watcher) reload(err error) []*MSG {
+ log.Println("Falling back to a full policy reload:", err)
+ return []*MSG{w.newMessage(Update)}
+}
- // Marshal and send message
+// ruleMessages builds one message per section and policy type.
+func (w *Watcher) ruleMessages(rules []rule, single, batch UpdateType) []*MSG {
+ groups := groupRules(rules)
+ msgs := make([]*MSG, 0, len(groups))
+
+ for _, group := range groups {
+ msg := w.newMessage(single)
+ msg.Sec = group[0].sec
+ msg.Ptype = group[0].ptype
+
+ if len(group) == 1 {
+ msg.NewRule = group[0].fields
+ } else {
+ msg.Method = batch
+ msg.NewRules = ruleFields(group)
+ }
+
+ msgs = append(msgs, msg)
+ }
+
+ return msgs
+}
+
+func (w *Watcher) newMessage(method UpdateType) *MSG {
+ return &MSG{Method: method, ID: w.localID}
+}
+
+// notify hands one message to the callback. Calls are serialized so a change
+// made of several messages, such as a rule replacement, is applied in order.
+func (w *Watcher) notify(callback func(string), msg *MSG) {
msgBytes, err := msg.MarshalBinary()
if err != nil {
log.Println("Error marshaling message:", err)
return
}
+ w.callbackMu.Lock()
+ defer w.callbackMu.Unlock()
+
callback(string(msgBytes))
}
diff --git a/watcher_test.go b/watcher_test.go
index 496a110..5715c65 100644
--- a/watcher_test.go
+++ b/watcher_test.go
@@ -263,7 +263,7 @@ func TestHandleEvent(t *testing.T) {
}
// Handle the event
- w.handleEvent(obj, "add")
+ w.handleEvent(eventAdd, nil, obj)
time.Sleep(100 * time.Millisecond)
@@ -318,7 +318,7 @@ func TestIgnoreSelf(t *testing.T) {
},
}
- w.handleEvent(obj, "add")
+ w.handleEvent(eventAdd, nil, obj)
time.Sleep(100 * time.Millisecond)
if callbackCalled {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
