1package app
2
3import (
4 "errors"
5 "fmt"
6)
7
8// applyProgress tracks whether the outputs of a DRM configuration are ready.
9// It belongs exclusively to the output loop. An instance is identified by its
10// Ready channel, which is distinct for each start of a connector.
11type applyProgress struct {
12 id uint64
13 active bool
14 required map[string]bool
15 waiting map[string]bool // stopped instances awaiting replacement or removal
16 ready map[string]<-chan error
17 completed map[<-chan error]bool
18 err error
19}
20
21// applyDecision reports whether the configuration is finished and its result.
22type applyDecision struct {
23 reply bool
24 err error
25}
26
27func (p *applyProgress) start(id uint64, required map[string]bool, stopping map[string]bool) applyDecision {
28 p.id, p.active, p.err = id, true, nil
29 p.required = required
30 p.waiting = map[string]bool{}
31 p.ready = map[string]<-chan error{}
32 for name := range stopping {
33 p.waiting[name] = true
34 }
35 return applyDecision{}
36}
37
38func (p *applyProgress) stopping(name string) {
39 if p.active {
40 p.waiting[name] = true
41 delete(p.ready, name)
42 }
43}
44
45func (p *applyProgress) started(name string, instance <-chan error) {
46 if p.active && p.required[name] && !p.completed[instance] && !p.waiting[name] {
47 p.ready[name] = instance
48 }
49}
50
51// readyEvent records a readiness result. current is the connector's running
52// instance: results of a stopped instance are ignored.
53func (p *applyProgress) readyEvent(name string, instance, current <-chan error, err error) applyDecision {
54 if instance != current {
55 return applyDecision{}
56 }
57 if p.completed == nil {
58 p.completed = map[<-chan error]bool{}
59 }
60 if err == nil {
61 p.completed[instance] = true
62 }
63 if p.active && p.ready[name] == instance {
64 delete(p.ready, name)
65 if err != nil {
66 return p.fail(fmt.Errorf("%s: %w", name, err))
67 }
68 } else if p.active && err != nil && p.required[name] && !p.waiting[name] {
69 return p.fail(fmt.Errorf("%s: %w", name, err))
70 }
71 return p.complete()
72}
73
74func (p *applyProgress) stopped(name string, instance <-chan error, restart bool, err error) applyDecision {
75 delete(p.completed, instance)
76 if !p.active {
77 return applyDecision{}
78 }
79 if p.ready[name] == instance {
80 delete(p.ready, name)
81 }
82 if err != nil {
83 return p.fail(fmt.Errorf("%s stopped: %w", name, err))
84 }
85 if restart && p.required[name] {
86 p.waiting[name] = false
87 return applyDecision{}
88 } else {
89 delete(p.waiting, name)
90 if p.required[name] {
91 return p.fail(fmt.Errorf("%s stopped before ready", name))
92 }
93 }
94 return p.complete()
95}
96
97func (p *applyProgress) scanned(running map[string]<-chan error) applyDecision {
98 if !p.active {
99 return applyDecision{}
100 }
101 for name := range p.required {
102 instance := running[name]
103 if instance == nil {
104 return p.fail(fmt.Errorf("%s failed to start", name))
105 }
106 if stopping, ok := p.waiting[name]; ok {
107 if stopping {
108 continue
109 }
110 delete(p.waiting, name)
111 }
112 if !p.completed[instance] {
113 p.ready[name] = instance
114 } else {
115 delete(p.ready, name)
116 }
117 }
118 return p.complete()
119}
120
121// scanError fails the active operation even if an old output was already ready.
122func (p *applyProgress) scanError(err error) applyDecision {
123 if !p.active || err == nil {
124 return applyDecision{}
125 }
126 return p.fail(err)
127}
128
129func (p *applyProgress) timeout(id uint64) applyDecision {
130 if !p.active || id != p.id {
131 return applyDecision{}
132 }
133 return p.fail(errTimeout)
134}
135
136func (p *applyProgress) fail(err error) applyDecision {
137 p.err = errors.Join(p.err, err)
138 p.active = false
139 return applyDecision{reply: true, err: p.err}
140}
141
142func (p *applyProgress) complete() applyDecision {
143 if !p.active || len(p.waiting) != 0 || len(p.ready) != 0 {
144 return applyDecision{}
145 }
146 p.active = false
147 return applyDecision{reply: true, err: p.err}
148}