diff --git a/go/views/process_reader.go b/go/views/process_reader.go index 5115032f..3ac45459 100644 --- a/go/views/process_reader.go +++ b/go/views/process_reader.go @@ -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() diff --git a/go/views/views_test.go b/go/views/views_test.go index 07bdd0ac..e0fcc69a 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -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" {