RN12: a process reader hears a seat's event with no picture on the event's one subscription
AdoptEvent hands each message of the event to its handler beside the adopted views, and makes the subscription at once when no view made it, so the runner hears a run's candidate revision even when it follows no watched branch. Builds: design 47 G A-185, A-199, A-193, §532 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -179,6 +179,32 @@ func (r *ProcessReader) Adopt(f *Follower, seat, event string, hear func(context
|
||||
return p, nil
|
||||
}
|
||||
|
||||
// AdoptEvent hears a seat's event on the one subscription the followers adopted on that event share, with no picture and no
|
||||
// first view (design 47 G A-185, A-199): hear is handed each message of the event beside every adopted follower's handler,
|
||||
// and the message is acknowledged. The subscription is made at once when no follower made it, so a part that follows no
|
||||
// view of the seat still hears the event; adopted before Start, it is made at Start.
|
||||
func (r *ProcessReader) AdoptEvent(seat, event string, hear func(context.Context, []byte) error) error {
|
||||
if seat == "" || event == "" || hear == nil {
|
||||
return errors.New("views: an adopted event names its seat, its event and its handler")
|
||||
}
|
||||
subject := "mesh.seat." + seat + ".event." + event
|
||||
r.mu.Lock()
|
||||
if r.adopted == nil {
|
||||
r.adopted = map[string][]adoptedHear{}
|
||||
}
|
||||
fresh := len(r.adopted[subject]) == 0
|
||||
r.adopted[subject] = append(r.adopted[subject], adoptedHear{hear: hear})
|
||||
hub, handle := r.hub, r.handle
|
||||
r.mu.Unlock()
|
||||
if fresh && hub != nil {
|
||||
// one subscription per event, with every further part (design 47 G §532)
|
||||
if err := hub.Handle(subject+".>", handle); err != nil {
|
||||
r.logf("views: %s's event %s is not heard (%v)", r.Module, subject, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// adopt is Adopt's part under the lock; subject is the event's subject when no follower adopted it before.
|
||||
func (r *ProcessReader) adopt(f *Follower, seat, event string, hear func(context.Context, []byte) error) (*Picture, string, error) {
|
||||
r.mu.Lock()
|
||||
|
||||
@@ -800,6 +800,26 @@ func subViews(t *testing.T) {
|
||||
if len(a) != 1 || len(b) != 1 {
|
||||
t.Errorf("two adopted views on one process reader heard %v and %v", a, b)
|
||||
}
|
||||
// A-185, A-199: an event adopted with no picture is heard once on the same subscription, beside the views; a reader
|
||||
// following no view hears it too.
|
||||
var hooked []string
|
||||
hook := func(_ context.Context, d []byte) error { hooked = append(hooked, string(d)); return nil }
|
||||
if err := prr.AdoptEvent("s", "revision-said", hook); err != nil || prr.AdoptEvent("s", "", hook) == nil {
|
||||
t.Fatalf("AdoptEvent: %v, and one naming no event was taken", err)
|
||||
}
|
||||
prr.Hear(context.Background())(&seatMsg{subject: "mesh.seat.s.event.revision-said.main", data: []byte(`{"revision":8}`)})
|
||||
if len(a) != 2 || len(b) != 2 || len(hooked) != 1 || len(prr.Followers()) != 2 {
|
||||
t.Errorf("beside two views the adopted event heard %v (views %v, %v)", hooked, a, b)
|
||||
}
|
||||
alone := &ProcessReader{Module: "reader", Now: now}
|
||||
if err := alone.AdoptEvent("s", "revision-said", hook); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
am := &seatMsg{subject: "mesh.seat.s.event.revision-said.next", data: []byte(`{"revision":9}`)}
|
||||
alone.Hear(context.Background())(am)
|
||||
if len(hooked) != 2 || !am.acked {
|
||||
t.Errorf("a reader following no view heard %v, acked %v", hooked, am.acked)
|
||||
}
|
||||
pf.WaitFor(context.Background(), "4")
|
||||
ended <- true
|
||||
if fd := pfs.next(t); fd.Kind != ViewWaitsKind || fd.ID != "reader.s.revisions" {
|
||||
|
||||
Reference in New Issue
Block a user