1package capture
2
3import (
4 "context"
5 "errors"
6 "time"
7
8 "github.com/bnema/neferwl/internal/ports"
9)
10
11// Reasons an output owner fails a request with.
12var (
13 ErrOutputStopped = errors.New("output stopped")
14 ErrOutputOff = errors.New("output off")
15 ErrSessionProtected = errors.New("session protected")
16 ErrBatchFull = errors.New("output capture batch full")
17 ErrEpochChanged = errors.New("security epoch changed")
18)
19
20// Requests are the capture requests an output owner collected for its next
21// frame, in fixed storage (MaxRequests): no allocation. Its owner goroutine
22// is the only user. List is the live slice the pipeline calls take and
23// return; assign their result back to it.
24type Requests struct {
25 storage [MaxRequests]ports.CaptureRequest
26 List []ports.CaptureRequest
27 ctx context.Context
28 replies chan<- ports.CaptureDone
29}
30
31// Init binds the requests to the owner's context and reply channel. It must
32// run before any other method.
33func (r *Requests) Init(ctx context.Context, replies chan<- ports.CaptureDone) {
34 r.List, r.ctx, r.replies = r.storage[:0], ctx, replies
35}
36
37// Len is the number of requests held.
38func (r *Requests) Len() int { return len(r.List) }
39
40// Admit takes q for the next frame, or fails it with refuse when that is not
41// nil, or when the batch is full. It reports whether q needs a new frame: its
42// indicator is already on the current scene.
43func (r *Requests) Admit(q ports.CaptureRequest, refuse error, scene ports.Scene, haveScene bool) (redraw bool) {
44 if refuse == nil && len(r.List) == cap(r.List) {
45 refuse = ErrBatchFull
46 }
47 if refuse != nil {
48 Fail(r.ctx, q, refuse, r.replies)
49 return false
50 }
51 q.Since = time.Now()
52 r.List = append(r.List, q)
53 // Its indicator may already be on screen; else it waits for the scene
54 // that shows it.
55 return haveScene && IndicatorShown(scene, q)
56}
57
58// FailPending fails every request not yet handed to the capture worker and
59// empties the list.
60func (r *Requests) FailPending(err error) {
61 for _, q := range r.List {
62 if !Handed(q) {
63 Fail(r.ctx, q, err, r.replies)
64 }
65 }
66 r.Reset()
67}
68
69// Reset empties the list without answering: its requests were answered or
70// handed over.
71func (r *Requests) Reset() {
72 clear(r.List)
73 r.List = r.storage[:0]
74}
75
76// Stop fails what the owner still holds, then what is queued on incoming,
77// with ErrOutputStopped. It returns once incoming is empty or closed.
78func (r *Requests) Stop(incoming <-chan ports.CaptureRequest) {
79 for _, q := range r.List {
80 if !Handed(q) {
81 Fail(r.ctx, q, ErrOutputStopped, r.replies)
82 }
83 }
84 r.Reset()
85 for {
86 select {
87 case q, ok := <-incoming:
88 if !ok {
89 return
90 }
91 Fail(r.ctx, q, ErrOutputStopped, r.replies)
92 default:
93 return
94 }
95 }
96}