diff --git a/modules/pacman/cmd/pacman-tools/keeper.go b/modules/pacman/cmd/pacman-tools/keeper.go index a7ebb46e..ede386f3 100644 --- a/modules/pacman/cmd/pacman-tools/keeper.go +++ b/modules/pacman/cmd/pacman-tools/keeper.go @@ -77,10 +77,6 @@ type Store interface { Watch(key string, handle func(stdio.StateChange) error) error } -// Emitter emits one event with its body; id is the event's id, the bus's dedup id, so emitting it again after a -// failure the bus did not report is dropped. -type Emitter func(event string, body any, id string) error - // Clock is the time as the keeper sees it. type Clock interface { Now() time.Time @@ -154,7 +150,9 @@ type Keeper struct { Config KeeperConfig Machine *Machine State func(name string) Store - Emit Emitter + // Events emits the keeper's events; package.changed carries the id `installed..`, the bus's + // dedup id, so the same change said twice is dropped. + Events EventSink Clock Clock Watcher Watcher // Locked says pacman's lock is held now: a released lock followed at once by another transaction is read @@ -334,7 +332,8 @@ func (k *Keeper) refreshInstalled(at time.Time) *PackageChanged { return nil } ev := &PackageChanged{Node: k.Node, At: at.UTC(), Added: c.Added, Removed: c.Removed, Changed: c.Changed} - if err := k.Emit(EventPackageChanged, *ev, fmt.Sprintf("%s.%s.%d", StateInstalled, k.Node, rev)); err != nil { + id := fmt.Sprintf("%s.%s.%d", StateInstalled, k.Node, rev) + if err := k.Events.Emit(EventPackageChanged, *ev, stdio.EmitOptions{MessageID: id}); err != nil { k.logf("[pacman] keeper: emitting %s for installed.%s at revision %d: %v", EventPackageChanged, k.Node, rev, err) } return ev @@ -521,8 +520,6 @@ func runKeeper(ctx context.Context, account process.Account, k *Keeper) error { k.Node = account.Node } k.State = func(name string) Store { return processStore{conn.State(name)} } - k.Emit = func(event string, body any, id string) error { - return conn.Emit(event, body, stdio.EmitOptions{MessageID: id}) - } + k.Events = conn return k.Run(ctx) } diff --git a/modules/pacman/cmd/pacman-tools/keeper_test.go b/modules/pacman/cmd/pacman-tools/keeper_test.go index 0e027cb7..21ddf094 100644 --- a/modules/pacman/cmd/pacman-tools/keeper_test.go +++ b/modules/pacman/cmd/pacman-tools/keeper_test.go @@ -202,9 +202,13 @@ type emittedEvent struct { id string } -func (e *emitted) Emit(name string, body any, id string) error { +func (e *emitted) Emit(name string, body any, opts ...stdio.EmitOptions) error { e.mu.Lock() defer e.mu.Unlock() + id := "" + for _, o := range opts { + id = o.MessageID + } e.events = append(e.events, emittedEvent{name, body, id}) return nil } @@ -318,7 +322,7 @@ func newTestKeeper(db *pacmanDB) (*Keeper, *fakeStates, *emitted, *fakeClock, fa states, ev, clock := &fakeStates{}, &emitted{}, &fakeClock{now: t0} w := fakeWatcher{make(chan Signal)} m := machine(db.runner(), 1000) - k := &Keeper{Node: "node-a", Machine: m, State: states.State, Emit: ev.Emit, Clock: clock, Watcher: w, + k := &Keeper{Node: "node-a", Machine: m, State: states.State, Events: ev, Clock: clock, Watcher: w, Logf: func(string, ...any) {}} return k, states, ev, clock, w }