DLV-F3r sdk: a reader hears a views event by its data, and a gap is asked as a gap (A7's request)
views.HearEvent routes one event given as its data, for the tool runner, which takes the owners' events by its own path; Follower.Gap asks once more without counting it by hand. Hear is HearEvent on a consumer's message. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
+27
-17
@@ -317,31 +317,41 @@ func (f *Follower) Apply(ev Event) (gap bool, why string) {
|
||||
// that follower's first view once more on its own goroutine, said in the log. A message it cannot read is terminated.
|
||||
func Hear(ctx context.Context, followers map[string]*Follower, logf func(string, ...any)) func(jetstream.Msg) {
|
||||
return func(m jetstream.Msg) {
|
||||
ev, err := DecodeEvent(m.Data())
|
||||
if err != nil {
|
||||
if err := HearEvent(ctx, followers, m.Data(), logf); err != nil {
|
||||
logf("views: %s is not a views event, dropped: %v", m.Subject(), err)
|
||||
_ = m.Term()
|
||||
return
|
||||
}
|
||||
f := followers[ev.State]
|
||||
if f == nil {
|
||||
_ = m.Ack() // a state this reader does not follow
|
||||
return
|
||||
}
|
||||
if gap, why := f.Apply(ev); gap {
|
||||
logf("views: a gap in %s (%s): its first view %s is asked once more", ev.State, why, f.Tool)
|
||||
go func() {
|
||||
f.mu.Lock()
|
||||
f.init()
|
||||
f.st.Why = "a gap: " + why
|
||||
f.mu.Unlock()
|
||||
_ = f.askUntil(ctx, "a gap: "+why, false)
|
||||
}()
|
||||
}
|
||||
_ = m.Ack()
|
||||
}
|
||||
}
|
||||
|
||||
// HearEvent is Hear for one event given as its data, for a reader that takes the owner's events by another path than
|
||||
// a consumer of its own (the tool runner's): routed to the follower of its state, an event of a state it does not
|
||||
// follow passed over, and a gap asked once more on its own goroutine as a gap, never as a hand ask (Gap). Data that is
|
||||
// not a views event is answered as an error, and nothing is asked.
|
||||
func HearEvent(ctx context.Context, followers map[string]*Follower, data []byte, logf func(string, ...any)) error {
|
||||
ev, err := DecodeEvent(data)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
f := followers[ev.State]
|
||||
if f == nil {
|
||||
return nil // a state this reader does not follow
|
||||
}
|
||||
if gap, why := f.Apply(ev); gap {
|
||||
logf("views: a gap in %s (%s): its first view %s is asked once more", ev.State, why, f.Tool)
|
||||
go func() { _ = f.Gap(ctx, why) }()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Gap asks the first view once more for a gap the reader saw, until answered or ctx ends: counted among its asks,
|
||||
// never among its asks by hand.
|
||||
func (f *Follower) Gap(ctx context.Context, why string) error {
|
||||
return f.askUntil(ctx, "a gap: "+why, false)
|
||||
}
|
||||
|
||||
// AskVia is a Follower's Ask through a process's call of the owner's tool (process.Conn.Call), decoding its answer.
|
||||
func AskVia(call func(key string, body any) (json.RawMessage, error), tool string) func(context.Context, string) (Answer, error) {
|
||||
return func(_ context.Context, after string) (Answer, error) {
|
||||
|
||||
@@ -188,6 +188,30 @@ func TestViews(t *testing.T) {
|
||||
t.Errorf("d = %s, %v at %d", v, ok, f.Status().Revision)
|
||||
}
|
||||
})
|
||||
t.Run("an event heard by its data asks a gap once more as a gap, never by hand", func(t *testing.T) {
|
||||
asked := make(chan struct{}, 4)
|
||||
f, _ := follower(func(ctx context.Context, after string) (Answer, error) {
|
||||
asked <- struct{}{}
|
||||
return answer(ctx, after)
|
||||
})
|
||||
_ = f.Start(context.Background())
|
||||
<-asked
|
||||
raw, _ := ev(12, "a", `{}`).Encode()
|
||||
if err := HearEvent(context.Background(), map[string]*Follower{"modules": f}, raw, t.Logf); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
select {
|
||||
case <-asked:
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("a gap heard by its data asked nothing")
|
||||
}
|
||||
if s := f.Status(); s.Asks != 2 || s.ByHand != 0 {
|
||||
t.Errorf("a gap was counted by hand: %+v", s)
|
||||
}
|
||||
if err := HearEvent(context.Background(), nil, []byte("not an event"), t.Logf); err == nil {
|
||||
t.Error("data that is no views event was taken")
|
||||
}
|
||||
})
|
||||
t.Run("an event names its state on its subject, and decodes as it was encoded", func(t *testing.T) {
|
||||
if s := StateOfEvent("mesh.stream.mesh-delivery-indexer.event.views.config-candidates"); s != "config-candidates" {
|
||||
t.Errorf("state of subject = %q", s)
|
||||
|
||||
Reference in New Issue
Block a user