feed: inline handlers, remove FeedBlockingRead wrapper
This commit is contained in:
parent
721b49ffa2
commit
1537e35685
|
|
@ -680,19 +680,11 @@ func (ag *Agent) FeedWrite(data []byte) {
|
||||||
ag.notifyChange()
|
ag.notifyChange()
|
||||||
}
|
}
|
||||||
|
|
||||||
// FeedBlockingRead blocks until feed data differs from base.
|
// FeedGet returns the current feed data.
|
||||||
// Returns the data and a nextBase for the caller to use on the next call.
|
func (ag *Agent) FeedGet() []byte { return ag.feed.Get() }
|
||||||
// This is the same function used by both the 9P BlockOnce handler and
|
|
||||||
// the internal consumer goroutine.
|
// FeedHash returns the current feed content hash.
|
||||||
func (ag *Agent) FeedBlockingRead(ctx context.Context, base string) ([]byte, string, error) {
|
func (ag *Agent) FeedHash() string { return ag.feed.Hash() }
|
||||||
_, ok := ag.WaitChange(ctx, WatchFeed, base)
|
|
||||||
if !ok {
|
|
||||||
return nil, base, ctx.Err()
|
|
||||||
}
|
|
||||||
data := ag.feed.Get()
|
|
||||||
hash := ag.feed.Hash()
|
|
||||||
return data, hash, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// StartFeedConsumer spawns a goroutine that blocks on feed changes and
|
// StartFeedConsumer spawns a goroutine that blocks on feed changes and
|
||||||
// submits new data as prompts. Exits when ctx is cancelled.
|
// submits new data as prompts. Exits when ctx is cancelled.
|
||||||
|
|
@ -700,14 +692,15 @@ func (ag *Agent) StartFeedConsumer(ctx context.Context) {
|
||||||
go func() {
|
go func() {
|
||||||
var base string
|
var base string
|
||||||
for {
|
for {
|
||||||
data, nextBase, err := ag.FeedBlockingRead(ctx, base)
|
_, ok := ag.WaitChange(ctx, WatchFeed, base)
|
||||||
if err != nil {
|
if !ok {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
data := ag.feed.Get()
|
||||||
|
base = ag.feed.Hash()
|
||||||
if len(data) == 0 {
|
if len(data) == 0 {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
base = nextBase
|
|
||||||
ag.Submit(ctx, string(data))
|
ag.Submit(ctx, string(data))
|
||||||
ag.EnsureTrailingNewline()
|
ag.EnsureTrailingNewline()
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -397,24 +397,22 @@ func buildAgentChildren(a *agent.Agent, s *session.Session) []virtfs.FsNodeDecl
|
||||||
if len(data) == 0 {
|
if len(data) == 0 {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
// Skip NUL-only input.
|
|
||||||
allNul := true
|
|
||||||
for _, b := range data {
|
for _, b := range data {
|
||||||
if b != 0 {
|
if b != 0 {
|
||||||
allNul = false
|
a.FeedWrite(data)
|
||||||
break
|
return nil
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if allNul {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
a.FeedWrite(data)
|
|
||||||
return nil
|
return nil
|
||||||
}),
|
}),
|
||||||
virtfs.BlockOnce(func(cctx context.Context, base string) ([]byte, string, error) {
|
virtfs.BlockOnce(func(cctx context.Context, base string) ([]byte, string, error) {
|
||||||
merged, cancel := mergeCtx(s.Ctx, cctx)
|
merged, cancel := mergeCtx(s.Ctx, cctx)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
return a.FeedBlockingRead(merged, base)
|
_, ok := a.WaitChange(merged, agent.WatchFeed, base)
|
||||||
|
if !ok {
|
||||||
|
return nil, base, merged.Err()
|
||||||
|
}
|
||||||
|
return a.FeedGet(), a.FeedHash(), nil
|
||||||
}),
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("chat", 0444,
|
virtfs.FileNode("chat", 0444,
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue