refactor(wayland): detach content feeds by handle instead of name
contentLink now carries its server and implements ports.ContentReturn itself, replacing the contentReturn wrapper. Server.DetachOutput takes the handle AttachOutput returned and removes that link only, so a stale handle can never remove a newer link of the same name. Attaching a name that already has a link still replaces it. app stores the handle per running output and detaches with it, which removes the ordering caveat from the contentAttacher doc. The mock is regenerated. A wayland test attaches A twice and detaches the old handle: the new link still receives content.
- commit
- ba55b3c4f4c4f1442b854576f567cc296afb8c64
- author
- brice · · 2026-10-14 03:37 UTC
- committer
- brice · · 2026-10-14 04:34 UTC
- parent
- 3935f90
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
5 files · +109 -55
M internal/adapters/wayland/contentlink.go +44 -29
| @@ -19,7 +19,12 @@ const batchItems = 8 | |||
| 19 | 19 | // goroutine owns cur and the sending side of in; the output owns the batch it | |
| 20 | 20 | // received until it returns it. free holds the batches nobody uses: it has | |
| 21 | 21 | // room for all of them, so Return never blocks. | |
| 22 | + | // | |
| 23 | + | // The link is also the ports.ContentReturn handed to the output, and the | |
| 24 | + | // handle DetachOutput takes: a detach removes this link and no other, even | |
| 25 | + | // when a later attach of the same name replaced it. | |
| 22 | 26 | type contentLink struct { | |
| 27 | + | s *Server | |
| 23 | 28 | name string | |
| 24 | 29 | in chan *ports.ContentBatch | |
| 25 | 30 | free chan *ports.ContentBatch | |
| @@ -28,24 +33,17 @@ type contentLink struct { | |||
| 28 | 33 | // full, or every batch is out). Return only wakes the display when it is | |
| 29 | 34 | // set, so a steady stream of frames costs the display no extra wake. | |
| 30 | 35 | pending atomic.Bool | |
| 31 | - | ret *contentReturn | |
| 32 | - | } | |
| 33 | - | ||
| 34 | - | // contentReturn is the ports.ContentReturn of one link. | |
| 35 | - | type contentReturn struct { | |
| 36 | - | s *Server | |
| 37 | - | l *contentLink | |
| 38 | 36 | } | |
| 39 | 37 | ||
| 40 | 38 | // Return gives a consumed (reset) batch back and, when contents wait for one, | |
| 41 | 39 | // has the display goroutine send them. Safe from any goroutine, also after | |
| 42 | 40 | // the link was detached: the batch only goes back to its own link. | |
| 43 | - | func (r *contentReturn) Return(b *ports.ContentBatch) { | |
| 44 | - | r.l.free <- b // never full: the link owns batchesPerLink batches | |
| 41 | + | func (l *contentLink) Return(b *ports.ContentBatch) { | |
| 42 | + | l.free <- b // never full: the link owns batchesPerLink batches | |
| 45 | 43 | // The display stores pending before it looks at free once more, so | |
| 46 | 44 | // either it sees this batch or this load sees pending. | |
| 47 | - | if r.l.pending.Load() && r.s.flushPosted.CompareAndSwap(false, true) { | |
| 48 | - | r.s.display.Post(r.s.flushLinksFn) | |
| 45 | + | if l.pending.Load() && l.s.flushPosted.CompareAndSwap(false, true) { | |
| 46 | + | l.s.display.Post(l.s.flushLinksFn) | |
| 49 | 47 | } | |
| 50 | 48 | } | |
| 51 | 49 | ||
| @@ -107,17 +105,24 @@ func (s *Server) flushLinks() { | |||
| 107 | 105 | } | |
| 108 | 106 | ||
| 109 | 107 | // attachOutput starts feeding the output called name; a link of the same name | |
| 110 | - | // is replaced. The first batch holds the latest content of every live window, | |
| 108 | + | // is replaced (dropped from the feed: its handle then detaches nothing). The | |
| 109 | + | // returned ContentReturn is the link, and the handle for detachOutput. The | |
| 110 | + | // first batch holds the latest content of every live window, | |
| 111 | 111 | // so the output starts with all of them drawn. Display goroutine, or before | |
| 112 | 112 | // Run. | |
| 113 | 113 | func (s *Server) attachOutput(name string) (<-chan *ports.ContentBatch, ports.ContentReturn) { | |
| 114 | - | s.detachOutput(name) | |
| 114 | + | for i, old := range s.links { | |
| 115 | + | if old.name == name { | |
| 116 | + | s.links = slices.Delete(s.links, i, i+1) | |
| 117 | + | break | |
| 118 | + | } | |
| 119 | + | } | |
| 115 | 120 | l := &contentLink{ | |
| 121 | + | s: s, | |
| 116 | 122 | name: name, | |
| 117 | 123 | in: make(chan *ports.ContentBatch, 1), | |
| 118 | 124 | free: make(chan *ports.ContentBatch, batchesPerLink), | |
| 119 | 125 | } | |
| 120 | - | l.ret = &contentReturn{s: s, l: l} | |
| 121 | 126 | for range batchesPerLink - 1 { | |
| 122 | 127 | l.free <- &ports.ContentBatch{Items: make([]ports.SurfaceContent, 0, batchItems)} | |
| 123 | 128 | } | |
| @@ -127,25 +132,30 @@ func (s *Server) attachOutput(name string) (<-chan *ports.ContentBatch, ports.Co | |||
| 127 | 132 | } | |
| 128 | 133 | s.links = append(s.links, l) | |
| 129 | 134 | s.flushLink(l) | |
| 130 | - | return l.in, l.ret | |
| 135 | + | return l.in, l | |
| 131 | 136 | } | |
| 132 | 137 | ||
| 133 | - | // detachOutput stops feeding the output called name. Display goroutine, or | |
| 134 | - | // before Run. The output may still Return its last batch. | |
| 135 | - | func (s *Server) detachOutput(name string) { | |
| 136 | - | for i, l := range s.links { | |
| 137 | - | if l.name == name { | |
| 138 | - | s.links = slices.Delete(s.links, i, i+1) | |
| 139 | - | return | |
| 140 | - | } | |
| 138 | + | // detachOutput stops feeding the link ret, the handle attachOutput returned. | |
| 139 | + | // A handle whose link was replaced by a later attach of its name, or that is | |
| 140 | + | // not a link, detaches nothing: a newer link is never removed. Display | |
| 141 | + | // goroutine, or before Run. The output may still Return its last batch. | |
| 142 | + | func (s *Server) detachOutput(ret ports.ContentReturn) { | |
| 143 | + | l, ok := ret.(*contentLink) | |
| 144 | + | if !ok { | |
| 145 | + | return | |
| 146 | + | } | |
| 147 | + | if i := slices.Index(s.links, l); i >= 0 { | |
| 148 | + | s.links = slices.Delete(s.links, i, i+1) | |
| 141 | 149 | } | |
| 142 | 150 | } | |
| 143 | 151 | ||
| 144 | 152 | // AttachOutput starts feeding the output called name with the contents of | |
| 145 | 153 | // the windows, as batches on the returned channel, starting with the current | |
| 146 | 154 | // content of every window. The output hands each batch back, reset, to the | |
| 147 | - | // returned ContentReturn. Call it from outside the display goroutine. It | |
| 148 | - | // returns a nil channel when the server is stopped. | |
| 155 | + | // returned ContentReturn, which is also the handle for DetachOutput. | |
| 156 | + | // Attaching a name that already has a feed replaces that feed. Call it from | |
| 157 | + | // outside the display goroutine. It returns a nil channel and handle when | |
| 158 | + | // the server is stopped. | |
| 149 | 159 | func (s *Server) AttachOutput(name string) (<-chan *ports.ContentBatch, ports.ContentReturn) { | |
| 150 | 160 | var in <-chan *ports.ContentBatch | |
| 151 | 161 | var ret ports.ContentReturn | |
| @@ -155,8 +165,13 @@ func (s *Server) AttachOutput(name string) (<-chan *ports.ContentBatch, ports.Co | |||
| 155 | 165 | return in, ret | |
| 156 | 166 | } | |
| 157 | 167 | ||
| 158 | - | // DetachOutput stops the feed of the output called name. Call it from outside | |
| 159 | - | // the display goroutine, once the output no longer reads its channel. | |
| 160 | - | func (s *Server) DetachOutput(name string) { | |
| 161 | - | s.display.Do(func() { s.detachOutput(name) }) | |
| 168 | + | // DetachOutput stops the feed ret, returned by AttachOutput. Only that feed | |
| 169 | + | // stops: if a later AttachOutput of the same name replaced it, the newer | |
| 170 | + | // feed keeps running. Call it from outside the display goroutine, once the | |
| 171 | + | // output no longer reads its channel. A nil ret does nothing. | |
| 172 | + | func (s *Server) DetachOutput(ret ports.ContentReturn) { | |
| 173 | + | if ret == nil { | |
| 174 | + | return | |
| 175 | + | } | |
| 176 | + | s.display.Do(func() { s.detachOutput(ret) }) | |
| 162 | 177 | } | |
| @@ -19,7 +19,12 @@ const batchItems = 8 | |||||
| 19 | // goroutine owns cur and the sending side of in; the output owns the batch it | 19 | // goroutine owns cur and the sending side of in; the output owns the batch it | ||
| 20 | // received until it returns it. free holds the batches nobody uses: it has | 20 | // received until it returns it. free holds the batches nobody uses: it has | ||
| 21 | // room for all of them, so Return never blocks. | 21 | // room for all of them, so Return never blocks. | ||
| 22 | + | // | |||
| 23 | + | // The link is also the ports.ContentReturn handed to the output, and the | |||
| 24 | + | // handle DetachOutput takes: a detach removes this link and no other, even | |||
| 25 | + | // when a later attach of the same name replaced it. | |||
| 22 | type contentLink struct { | 26 | type contentLink struct { | ||
| 27 | + | s *Server | |||
| 23 | name string | 28 | name string | ||
| 24 | in chan *ports.ContentBatch | 29 | in chan *ports.ContentBatch | ||
| 25 | free chan *ports.ContentBatch | 30 | free chan *ports.ContentBatch | ||
| @@ -28,24 +33,17 @@ type contentLink struct { | |||||
| 28 | // full, or every batch is out). Return only wakes the display when it is | 33 | // full, or every batch is out). Return only wakes the display when it is | ||
| 29 | // set, so a steady stream of frames costs the display no extra wake. | 34 | // set, so a steady stream of frames costs the display no extra wake. | ||
| 30 | pending atomic.Bool | 35 | pending atomic.Bool | ||
| 31 | - | ret *contentReturn | |||
| 32 | - | } | |||
| 33 | - | ||||
| 34 | - | // contentReturn is the ports.ContentReturn of one link. | |||
| 35 | - | type contentReturn struct { | |||
| 36 | - | s *Server | |||
| 37 | - | l *contentLink | |||
| 38 | } | 36 | } | ||
| 39 | 37 | ||||
| 40 | // Return gives a consumed (reset) batch back and, when contents wait for one, | 38 | // Return gives a consumed (reset) batch back and, when contents wait for one, | ||
| 41 | // has the display goroutine send them. Safe from any goroutine, also after | 39 | // has the display goroutine send them. Safe from any goroutine, also after | ||
| 42 | // the link was detached: the batch only goes back to its own link. | 40 | // the link was detached: the batch only goes back to its own link. | ||
| 43 | - | func (r *contentReturn) Return(b *ports.ContentBatch) { | 41 | + | func (l *contentLink) Return(b *ports.ContentBatch) { |
| 44 | - | r.l.free <- b // never full: the link owns batchesPerLink batches | 42 | + | l.free <- b // never full: the link owns batchesPerLink batches |
| 45 | // The display stores pending before it looks at free once more, so | 43 | // The display stores pending before it looks at free once more, so | ||
| 46 | // either it sees this batch or this load sees pending. | 44 | // either it sees this batch or this load sees pending. | ||
| 47 | - | if r.l.pending.Load() && r.s.flushPosted.CompareAndSwap(false, true) { | 45 | + | if l.pending.Load() && l.s.flushPosted.CompareAndSwap(false, true) { |
| 48 | - | r.s.display.Post(r.s.flushLinksFn) | 46 | + | l.s.display.Post(l.s.flushLinksFn) |
| 49 | } | 47 | } | ||
| 50 | } | 48 | } | ||
| 51 | 49 | ||||
| @@ -107,17 +105,24 @@ func (s *Server) flushLinks() { | |||||
| 107 | } | 105 | } | ||
| 108 | 106 | ||||
| 109 | // attachOutput starts feeding the output called name; a link of the same name | 107 | // attachOutput starts feeding the output called name; a link of the same name | ||
| 110 | - | // is replaced. The first batch holds the latest content of every live window, | 108 | + | // is replaced (dropped from the feed: its handle then detaches nothing). The |
| 109 | + | // returned ContentReturn is the link, and the handle for detachOutput. The | |||
| 110 | + | // first batch holds the latest content of every live window, | |||
| 111 | // so the output starts with all of them drawn. Display goroutine, or before | 111 | // so the output starts with all of them drawn. Display goroutine, or before | ||
| 112 | // Run. | 112 | // Run. | ||
| 113 | func (s *Server) attachOutput(name string) (<-chan *ports.ContentBatch, ports.ContentReturn) { | 113 | func (s *Server) attachOutput(name string) (<-chan *ports.ContentBatch, ports.ContentReturn) { | ||
| 114 | - | s.detachOutput(name) | 114 | + | for i, old := range s.links { |
| 115 | + | if old.name == name { | |||
| 116 | + | s.links = slices.Delete(s.links, i, i+1) | |||
| 117 | + | break | |||
| 118 | + | } | |||
| 119 | + | } | |||
| 115 | l := &contentLink{ | 120 | l := &contentLink{ | ||
| 121 | + | s: s, | |||
| 116 | name: name, | 122 | name: name, | ||
| 117 | in: make(chan *ports.ContentBatch, 1), | 123 | in: make(chan *ports.ContentBatch, 1), | ||
| 118 | free: make(chan *ports.ContentBatch, batchesPerLink), | 124 | free: make(chan *ports.ContentBatch, batchesPerLink), | ||
| 119 | } | 125 | } | ||
| 120 | - | l.ret = &contentReturn{s: s, l: l} | |||
| 121 | for range batchesPerLink - 1 { | 126 | for range batchesPerLink - 1 { | ||
| 122 | l.free <- &ports.ContentBatch{Items: make([]ports.SurfaceContent, 0, batchItems)} | 127 | l.free <- &ports.ContentBatch{Items: make([]ports.SurfaceContent, 0, batchItems)} | ||
| 123 | } | 128 | } | ||
| @@ -127,25 +132,30 @@ func (s *Server) attachOutput(name string) (<-chan *ports.ContentBatch, ports.Co | |||||
| 127 | } | 132 | } | ||
| 128 | s.links = append(s.links, l) | 133 | s.links = append(s.links, l) | ||
| 129 | s.flushLink(l) | 134 | s.flushLink(l) | ||
| 130 | - | return l.in, l.ret | 135 | + | return l.in, l |
| 131 | } | 136 | } | ||
| 132 | 137 | ||||
| 133 | - | // detachOutput stops feeding the output called name. Display goroutine, or | 138 | + | // detachOutput stops feeding the link ret, the handle attachOutput returned. |
| 134 | - | // before Run. The output may still Return its last batch. | 139 | + | // A handle whose link was replaced by a later attach of its name, or that is |
| 135 | - | func (s *Server) detachOutput(name string) { | 140 | + | // not a link, detaches nothing: a newer link is never removed. Display |
| 136 | - | for i, l := range s.links { | 141 | + | // goroutine, or before Run. The output may still Return its last batch. |
| 137 | - | if l.name == name { | 142 | + | func (s *Server) detachOutput(ret ports.ContentReturn) { |
| 138 | - | s.links = slices.Delete(s.links, i, i+1) | 143 | + | l, ok := ret.(*contentLink) |
| 139 | - | return | 144 | + | if !ok { |
| 140 | - | } | 145 | + | return |
| 146 | + | } | |||
| 147 | + | if i := slices.Index(s.links, l); i >= 0 { | |||
| 148 | + | s.links = slices.Delete(s.links, i, i+1) | |||
| 141 | } | 149 | } | ||
| 142 | } | 150 | } | ||
| 143 | 151 | ||||
| 144 | // AttachOutput starts feeding the output called name with the contents of | 152 | // AttachOutput starts feeding the output called name with the contents of | ||
| 145 | // the windows, as batches on the returned channel, starting with the current | 153 | // the windows, as batches on the returned channel, starting with the current | ||
| 146 | // content of every window. The output hands each batch back, reset, to the | 154 | // content of every window. The output hands each batch back, reset, to the | ||
| 147 | - | // returned ContentReturn. Call it from outside the display goroutine. It | 155 | + | // returned ContentReturn, which is also the handle for DetachOutput. |
| 148 | - | // returns a nil channel when the server is stopped. | 156 | + | // Attaching a name that already has a feed replaces that feed. Call it from |
| 157 | + | // outside the display goroutine. It returns a nil channel and handle when | |||
| 158 | + | // the server is stopped. | |||
| 149 | func (s *Server) AttachOutput(name string) (<-chan *ports.ContentBatch, ports.ContentReturn) { | 159 | func (s *Server) AttachOutput(name string) (<-chan *ports.ContentBatch, ports.ContentReturn) { | ||
| 150 | var in <-chan *ports.ContentBatch | 160 | var in <-chan *ports.ContentBatch | ||
| 151 | var ret ports.ContentReturn | 161 | var ret ports.ContentReturn | ||
| @@ -155,8 +165,13 @@ func (s *Server) AttachOutput(name string) (<-chan *ports.ContentBatch, ports.Co | |||||
| 155 | return in, ret | 165 | return in, ret | ||
| 156 | } | 166 | } | ||
| 157 | 167 | ||||
| 158 | - | // DetachOutput stops the feed of the output called name. Call it from outside | 168 | + | // DetachOutput stops the feed ret, returned by AttachOutput. Only that feed |
| 159 | - | // the display goroutine, once the output no longer reads its channel. | 169 | + | // stops: if a later AttachOutput of the same name replaced it, the newer |
| 160 | - | func (s *Server) DetachOutput(name string) { | 170 | + | // feed keeps running. Call it from outside the display goroutine, once the |
| 161 | - | s.display.Do(func() { s.detachOutput(name) }) | 171 | + | // output no longer reads its channel. A nil ret does nothing. |
| 172 | + | func (s *Server) DetachOutput(ret ports.ContentReturn) { | |||
| 173 | + | if ret == nil { | |||
| 174 | + | return | |||
| 175 | + | } | |||
| 176 | + | s.display.Do(func() { s.detachOutput(ret) }) | |||
| 162 | } | 177 | } | ||
M internal/adapters/wayland/contentlink_test.go +20 -1
| @@ -165,7 +165,7 @@ func TestContentBatchDetach(t *testing.T) { | |||
| 165 | 165 | if len(a.Items) != 1 || len(b.Items) != 1 || a == b { | |
| 166 | 166 | t.Fatalf("batches %+v %+v", a.Items, b.Items) | |
| 167 | 167 | } | |
| 168 | - | s.DetachOutput("A") | |
| 168 | + | s.DetachOutput(retA) | |
| 169 | 169 | // Waiting contents of a detached link are never sent; Return is safe. | |
| 170 | 170 | a.Reset() | |
| 171 | 171 | retA.Return(a) | |
| @@ -186,6 +186,25 @@ func TestContentBatchDetach(t *testing.T) { | |||
| 186 | 186 | } | |
| 187 | 187 | } | |
| 188 | 188 | ||
| 189 | + | // Detaching a feed by its handle never removes a newer feed of the same | |
| 190 | + | // name: A attached again replaced the first link, and detaching the old | |
| 191 | + | // handle leaves the new one fed. | |
| 192 | + | func TestDetachOldHandleKeepsReplacedLink(t *testing.T) { | |
| 193 | + | s := linkServer(t) | |
| 194 | + | _, oldRet := s.AttachOutput("A") | |
| 195 | + | in, _ := s.AttachOutput("A") | |
| 196 | + | s.DetachOutput(oldRet) | |
| 197 | + | var links int | |
| 198 | + | s.display.Do(func() { links = len(s.links) }) | |
| 199 | + | if links != 1 { | |
| 200 | + | t.Fatalf("%d links, want the new one only", links) | |
| 201 | + | } | |
| 202 | + | s.publish(t, solid(1)) | |
| 203 | + | if b := nextBatch(t, in); len(b.Items) != 1 { | |
| 204 | + | t.Fatalf("new link batch %+v", b.Items) | |
| 205 | + | } | |
| 206 | + | } | |
| 207 | + | ||
| 189 | 208 | // Emit, send, consume and return a batch without allocating once the | |
| 190 | 209 | // batches have their capacity. The consumer runs inline, like an output that | |
| 191 | 210 | // keeps up. | |
| @@ -165,7 +165,7 @@ func TestContentBatchDetach(t *testing.T) { | |||||
| 165 | if len(a.Items) != 1 || len(b.Items) != 1 || a == b { | 165 | if len(a.Items) != 1 || len(b.Items) != 1 || a == b { | ||
| 166 | t.Fatalf("batches %+v %+v", a.Items, b.Items) | 166 | t.Fatalf("batches %+v %+v", a.Items, b.Items) | ||
| 167 | } | 167 | } | ||
| 168 | - | s.DetachOutput("A") | 168 | + | s.DetachOutput(retA) |
| 169 | // Waiting contents of a detached link are never sent; Return is safe. | 169 | // Waiting contents of a detached link are never sent; Return is safe. | ||
| 170 | a.Reset() | 170 | a.Reset() | ||
| 171 | retA.Return(a) | 171 | retA.Return(a) | ||
| @@ -186,6 +186,25 @@ func TestContentBatchDetach(t *testing.T) { | |||||
| 186 | } | 186 | } | ||
| 187 | } | 187 | } | ||
| 188 | 188 | ||||
| 189 | + | // Detaching a feed by its handle never removes a newer feed of the same | |||
| 190 | + | // name: A attached again replaced the first link, and detaching the old | |||
| 191 | + | // handle leaves the new one fed. | |||
| 192 | + | func TestDetachOldHandleKeepsReplacedLink(t *testing.T) { | |||
| 193 | + | s := linkServer(t) | |||
| 194 | + | _, oldRet := s.AttachOutput("A") | |||
| 195 | + | in, _ := s.AttachOutput("A") | |||
| 196 | + | s.DetachOutput(oldRet) | |||
| 197 | + | var links int | |||
| 198 | + | s.display.Do(func() { links = len(s.links) }) | |||
| 199 | + | if links != 1 { | |||
| 200 | + | t.Fatalf("%d links, want the new one only", links) | |||
| 201 | + | } | |||
| 202 | + | s.publish(t, solid(1)) | |||
| 203 | + | if b := nextBatch(t, in); len(b.Items) != 1 { | |||
| 204 | + | t.Fatalf("new link batch %+v", b.Items) | |||
| 205 | + | } | |||
| 206 | + | } | |||
| 207 | + | ||||
| 189 | // Emit, send, consume and return a batch without allocating once the | 208 | // Emit, send, consume and return a batch without allocating once the | ||
| 190 | // batches have their capacity. The consumer runs inline, like an output that | 209 | // batches have their capacity. The consumer runs inline, like an output that | ||
| 191 | // keeps up. | 210 | // keeps up. | ||
M internal/app/contentattacher_mock_test.go +9 -9
| @@ -110,8 +110,8 @@ func (_c *mockcontentAttacher_AttachOutput_Call) RunAndReturn(run func(name stri | |||
| 110 | 110 | } | |
| 111 | 111 | ||
| 112 | 112 | // DetachOutput provides a mock function for the type mockcontentAttacher | |
| 113 | - | func (_mock *mockcontentAttacher) DetachOutput(name string) { | |
| 114 | - | _mock.Called(name) | |
| 113 | + | func (_mock *mockcontentAttacher) DetachOutput(feed ports.ContentReturn) { | |
| 114 | + | _mock.Called(feed) | |
| 115 | 115 | return | |
| 116 | 116 | } | |
| 117 | 117 | ||
| @@ -121,16 +121,16 @@ type mockcontentAttacher_DetachOutput_Call struct { | |||
| 121 | 121 | } | |
| 122 | 122 | ||
| 123 | 123 | // DetachOutput is a helper method to define mock.On call | |
| 124 | - | // - name string | |
| 125 | - | func (_e *mockcontentAttacher_Expecter) DetachOutput(name any) *mockcontentAttacher_DetachOutput_Call { | |
| 126 | - | return &mockcontentAttacher_DetachOutput_Call{Call: _e.mock.On("DetachOutput", name)} | |
| 124 | + | // - feed ports.ContentReturn | |
| 125 | + | func (_e *mockcontentAttacher_Expecter) DetachOutput(feed any) *mockcontentAttacher_DetachOutput_Call { | |
| 126 | + | return &mockcontentAttacher_DetachOutput_Call{Call: _e.mock.On("DetachOutput", feed)} | |
| 127 | 127 | } | |
| 128 | 128 | ||
| 129 | - | func (_c *mockcontentAttacher_DetachOutput_Call) Run(run func(name string)) *mockcontentAttacher_DetachOutput_Call { | |
| 129 | + | func (_c *mockcontentAttacher_DetachOutput_Call) Run(run func(feed ports.ContentReturn)) *mockcontentAttacher_DetachOutput_Call { | |
| 130 | 130 | _c.Call.Run(func(args mock.Arguments) { | |
| 131 | - | var arg0 string | |
| 131 | + | var arg0 ports.ContentReturn | |
| 132 | 132 | if args[0] != nil { | |
| 133 | - | arg0 = args[0].(string) | |
| 133 | + | arg0 = args[0].(ports.ContentReturn) | |
| 134 | 134 | } | |
| 135 | 135 | run( | |
| 136 | 136 | arg0, | |
| @@ -144,7 +144,7 @@ func (_c *mockcontentAttacher_DetachOutput_Call) Return() *mockcontentAttacher_D | |||
| 144 | 144 | return _c | |
| 145 | 145 | } | |
| 146 | 146 | ||
| 147 | - | func (_c *mockcontentAttacher_DetachOutput_Call) RunAndReturn(run func(name string)) *mockcontentAttacher_DetachOutput_Call { | |
| 147 | + | func (_c *mockcontentAttacher_DetachOutput_Call) RunAndReturn(run func(feed ports.ContentReturn)) *mockcontentAttacher_DetachOutput_Call { | |
| 148 | 148 | _c.Run(run) | |
| 149 | 149 | return _c | |
| 150 | 150 | } | |
| @@ -110,8 +110,8 @@ func (_c *mockcontentAttacher_AttachOutput_Call) RunAndReturn(run func(name stri | |||||
| 110 | } | 110 | } | ||
| 111 | 111 | ||||
| 112 | // DetachOutput provides a mock function for the type mockcontentAttacher | 112 | // DetachOutput provides a mock function for the type mockcontentAttacher | ||
| 113 | - | func (_mock *mockcontentAttacher) DetachOutput(name string) { | 113 | + | func (_mock *mockcontentAttacher) DetachOutput(feed ports.ContentReturn) { |
| 114 | - | _mock.Called(name) | 114 | + | _mock.Called(feed) |
| 115 | return | 115 | return | ||
| 116 | } | 116 | } | ||
| 117 | 117 | ||||
| @@ -121,16 +121,16 @@ type mockcontentAttacher_DetachOutput_Call struct { | |||||
| 121 | } | 121 | } | ||
| 122 | 122 | ||||
| 123 | // DetachOutput is a helper method to define mock.On call | 123 | // DetachOutput is a helper method to define mock.On call | ||
| 124 | - | // - name string | 124 | + | // - feed ports.ContentReturn |
| 125 | - | func (_e *mockcontentAttacher_Expecter) DetachOutput(name any) *mockcontentAttacher_DetachOutput_Call { | 125 | + | func (_e *mockcontentAttacher_Expecter) DetachOutput(feed any) *mockcontentAttacher_DetachOutput_Call { |
| 126 | - | return &mockcontentAttacher_DetachOutput_Call{Call: _e.mock.On("DetachOutput", name)} | 126 | + | return &mockcontentAttacher_DetachOutput_Call{Call: _e.mock.On("DetachOutput", feed)} |
| 127 | } | 127 | } | ||
| 128 | 128 | ||||
| 129 | - | func (_c *mockcontentAttacher_DetachOutput_Call) Run(run func(name string)) *mockcontentAttacher_DetachOutput_Call { | 129 | + | func (_c *mockcontentAttacher_DetachOutput_Call) Run(run func(feed ports.ContentReturn)) *mockcontentAttacher_DetachOutput_Call { |
| 130 | _c.Call.Run(func(args mock.Arguments) { | 130 | _c.Call.Run(func(args mock.Arguments) { | ||
| 131 | - | var arg0 string | 131 | + | var arg0 ports.ContentReturn |
| 132 | if args[0] != nil { | 132 | if args[0] != nil { | ||
| 133 | - | arg0 = args[0].(string) | 133 | + | arg0 = args[0].(ports.ContentReturn) |
| 134 | } | 134 | } | ||
| 135 | run( | 135 | run( | ||
| 136 | arg0, | 136 | arg0, | ||
| @@ -144,7 +144,7 @@ func (_c *mockcontentAttacher_DetachOutput_Call) Return() *mockcontentAttacher_D | |||||
| 144 | return _c | 144 | return _c | ||
| 145 | } | 145 | } | ||
| 146 | 146 | ||||
| 147 | - | func (_c *mockcontentAttacher_DetachOutput_Call) RunAndReturn(run func(name string)) *mockcontentAttacher_DetachOutput_Call { | 147 | + | func (_c *mockcontentAttacher_DetachOutput_Call) RunAndReturn(run func(feed ports.ContentReturn)) *mockcontentAttacher_DetachOutput_Call { |
| 148 | _c.Run(run) | 148 | _c.Run(run) | ||
| 149 | return _c | 149 | return _c | ||
| 150 | } | 150 | } | ||
M internal/app/contentattacher_test.go +22 -7
| @@ -4,14 +4,16 @@ import ( | |||
| 4 | 4 | "context" | |
| 5 | 5 | "testing" | |
| 6 | 6 | ||
| 7 | + | "github.com/stretchr/testify/mock" | |
| 8 | + | ||
| 7 | 9 | portsmocks "github.com/bnema/neferwl/internal/mocks/ports" | |
| 8 | 10 | "github.com/bnema/neferwl/internal/ports" | |
| 9 | 11 | ) | |
| 10 | 12 | ||
| 11 | 13 | // Wayland feeds a started output its contents: Attach comes after the | |
| 12 | 14 | // output registered and got its security state, the output runs on what | |
| 13 | - | // Attach returned, Detach follows its end, and a restarted connector attaches | |
| 14 | - | // again. | |
| 15 | + | // Attach returned, Detach of that same handle follows its end, and a | |
| 16 | + | // restarted connector attaches again. | |
| 15 | 17 | func TestOutputSetAttachesAndDetachesContents(t *testing.T) { | |
| 16 | 18 | ctx, cancel := context.WithCancel(context.Background()) | |
| 17 | 19 | defer cancel() | |
| @@ -20,7 +22,7 @@ func TestOutputSetAttachesAndDetachesContents(t *testing.T) { | |||
| 20 | 22 | set.wireSecurity(outputChannels{securityEvents: events}) | |
| 21 | 23 | attacher := newMockcontentAttacher(t) | |
| 22 | 24 | set.attach = attacher | |
| 23 | - | ret := portsmocks.NewMockContentReturn(t) | |
| 25 | + | rets := []*portsmocks.MockContentReturn{portsmocks.NewMockContentReturn(t), portsmocks.NewMockContentReturn(t)} | |
| 24 | 26 | feed := make(chan *ports.ContentBatch, 1) | |
| 25 | 27 | var order []string | |
| 26 | 28 | type seen struct { | |
| @@ -34,21 +36,28 @@ func TestOutputSetAttachesAndDetachesContents(t *testing.T) { | |||
| 34 | 36 | <-ctx.Done() | |
| 35 | 37 | return nil | |
| 36 | 38 | } | |
| 39 | + | attaches := 0 | |
| 37 | 40 | attacher.EXPECT().AttachOutput("DP-1").RunAndReturn(func(string) (<-chan *ports.ContentBatch, ports.ContentReturn) { | |
| 38 | 41 | order = append(order, "attach") | |
| 42 | + | ret := rets[attaches] | |
| 43 | + | attaches++ | |
| 39 | 44 | // The output is registered with the security backend first. | |
| 40 | 45 | if len(events) != 1 { | |
| 41 | 46 | t.Errorf("attached with %d registrations queued, want 1", len(events)) | |
| 42 | 47 | } | |
| 43 | 48 | return feed, ret | |
| 44 | 49 | }).Times(2) | |
| 45 | - | attacher.EXPECT().DetachOutput("DP-1").Run(func(string) { order = append(order, "detach") }).Times(2) | |
| 50 | + | var detached []ports.ContentReturn | |
| 51 | + | attacher.EXPECT().DetachOutput(mock.Anything).Run(func(feed ports.ContentReturn) { | |
| 52 | + | order = append(order, "detach") | |
| 53 | + | detached = append(detached, feed) | |
| 54 | + | }).Times(2) | |
| 46 | 55 | for round := range 2 { | |
| 47 | 56 | if !set.start(ctx, "DP-1", run) { | |
| 48 | 57 | t.Fatal("start failed") | |
| 49 | 58 | } | |
| 50 | 59 | got := <-started | |
| 51 | - | if got.contents != (<-chan *ports.ContentBatch)(feed) || got.returns != ports.ContentReturn(ret) { | |
| 60 | + | if got.contents != (<-chan *ports.ContentBatch)(feed) || got.returns != ports.ContentReturn(rets[round]) { | |
| 52 | 61 | t.Fatalf("round %d: output got another feed", round) | |
| 53 | 62 | } | |
| 54 | 63 | <-events | |
| @@ -57,6 +66,11 @@ func TestOutputSetAttachesAndDetachesContents(t *testing.T) { | |||
| 57 | 66 | t.Fatal(err) | |
| 58 | 67 | } | |
| 59 | 68 | } | |
| 69 | + | for round := range detached { | |
| 70 | + | if detached[round] != ports.ContentReturn(rets[round]) { | |
| 71 | + | t.Fatalf("round %d: detached another feed than the one attached", round) | |
| 72 | + | } | |
| 73 | + | } | |
| 60 | 74 | want := []string{"attach", "detach", "attach", "detach"} | |
| 61 | 75 | if len(order) != len(want) { | |
| 62 | 76 | t.Fatalf("order %v, want %v", order, want) | |
| @@ -75,8 +89,9 @@ func TestOutputSetWaitDetachesContents(t *testing.T) { | |||
| 75 | 89 | set := newOutputSet(ctx, nil) | |
| 76 | 90 | attacher := newMockcontentAttacher(t) | |
| 77 | 91 | set.attach = attacher | |
| 78 | - | attacher.EXPECT().AttachOutput("A").Return(nil, nil).Once() | |
| 79 | - | attacher.EXPECT().DetachOutput("A").Return().Once() | |
| 92 | + | ret := portsmocks.NewMockContentReturn(t) | |
| 93 | + | attacher.EXPECT().AttachOutput("A").Return(nil, ret).Once() | |
| 94 | + | attacher.EXPECT().DetachOutput(ports.ContentReturn(ret)).Return().Once() | |
| 80 | 95 | set.start(ctx, "A", func(ctx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, _ ports.OutputInstance) error { | |
| 81 | 96 | <-ctx.Done() | |
| 82 | 97 | return nil | |
| @@ -4,14 +4,16 @@ import ( | |||||
| 4 | "context" | 4 | "context" | ||
| 5 | "testing" | 5 | "testing" | ||
| 6 | 6 | ||||
| 7 | + | "github.com/stretchr/testify/mock" | |||
| 8 | + | ||||
| 7 | portsmocks "github.com/bnema/neferwl/internal/mocks/ports" | 9 | portsmocks "github.com/bnema/neferwl/internal/mocks/ports" | ||
| 8 | "github.com/bnema/neferwl/internal/ports" | 10 | "github.com/bnema/neferwl/internal/ports" | ||
| 9 | ) | 11 | ) | ||
| 10 | 12 | ||||
| 11 | // Wayland feeds a started output its contents: Attach comes after the | 13 | // Wayland feeds a started output its contents: Attach comes after the | ||
| 12 | // output registered and got its security state, the output runs on what | 14 | // output registered and got its security state, the output runs on what | ||
| 13 | - | // Attach returned, Detach follows its end, and a restarted connector attaches | 15 | + | // Attach returned, Detach of that same handle follows its end, and a |
| 14 | - | // again. | 16 | + | // restarted connector attaches again. |
| 15 | func TestOutputSetAttachesAndDetachesContents(t *testing.T) { | 17 | func TestOutputSetAttachesAndDetachesContents(t *testing.T) { | ||
| 16 | ctx, cancel := context.WithCancel(context.Background()) | 18 | ctx, cancel := context.WithCancel(context.Background()) | ||
| 17 | defer cancel() | 19 | defer cancel() | ||
| @@ -20,7 +22,7 @@ func TestOutputSetAttachesAndDetachesContents(t *testing.T) { | |||||
| 20 | set.wireSecurity(outputChannels{securityEvents: events}) | 22 | set.wireSecurity(outputChannels{securityEvents: events}) | ||
| 21 | attacher := newMockcontentAttacher(t) | 23 | attacher := newMockcontentAttacher(t) | ||
| 22 | set.attach = attacher | 24 | set.attach = attacher | ||
| 23 | - | ret := portsmocks.NewMockContentReturn(t) | 25 | + | rets := []*portsmocks.MockContentReturn{portsmocks.NewMockContentReturn(t), portsmocks.NewMockContentReturn(t)} |
| 24 | feed := make(chan *ports.ContentBatch, 1) | 26 | feed := make(chan *ports.ContentBatch, 1) | ||
| 25 | var order []string | 27 | var order []string | ||
| 26 | type seen struct { | 28 | type seen struct { | ||
| @@ -34,21 +36,28 @@ func TestOutputSetAttachesAndDetachesContents(t *testing.T) { | |||||
| 34 | <-ctx.Done() | 36 | <-ctx.Done() | ||
| 35 | return nil | 37 | return nil | ||
| 36 | } | 38 | } | ||
| 39 | + | attaches := 0 | |||
| 37 | attacher.EXPECT().AttachOutput("DP-1").RunAndReturn(func(string) (<-chan *ports.ContentBatch, ports.ContentReturn) { | 40 | attacher.EXPECT().AttachOutput("DP-1").RunAndReturn(func(string) (<-chan *ports.ContentBatch, ports.ContentReturn) { | ||
| 38 | order = append(order, "attach") | 41 | order = append(order, "attach") | ||
| 42 | + | ret := rets[attaches] | |||
| 43 | + | attaches++ | |||
| 39 | // The output is registered with the security backend first. | 44 | // The output is registered with the security backend first. | ||
| 40 | if len(events) != 1 { | 45 | if len(events) != 1 { | ||
| 41 | t.Errorf("attached with %d registrations queued, want 1", len(events)) | 46 | t.Errorf("attached with %d registrations queued, want 1", len(events)) | ||
| 42 | } | 47 | } | ||
| 43 | return feed, ret | 48 | return feed, ret | ||
| 44 | }).Times(2) | 49 | }).Times(2) | ||
| 45 | - | attacher.EXPECT().DetachOutput("DP-1").Run(func(string) { order = append(order, "detach") }).Times(2) | 50 | + | var detached []ports.ContentReturn |
| 51 | + | attacher.EXPECT().DetachOutput(mock.Anything).Run(func(feed ports.ContentReturn) { | |||
| 52 | + | order = append(order, "detach") | |||
| 53 | + | detached = append(detached, feed) | |||
| 54 | + | }).Times(2) | |||
| 46 | for round := range 2 { | 55 | for round := range 2 { | ||
| 47 | if !set.start(ctx, "DP-1", run) { | 56 | if !set.start(ctx, "DP-1", run) { | ||
| 48 | t.Fatal("start failed") | 57 | t.Fatal("start failed") | ||
| 49 | } | 58 | } | ||
| 50 | got := <-started | 59 | got := <-started | ||
| 51 | - | if got.contents != (<-chan *ports.ContentBatch)(feed) || got.returns != ports.ContentReturn(ret) { | 60 | + | if got.contents != (<-chan *ports.ContentBatch)(feed) || got.returns != ports.ContentReturn(rets[round]) { |
| 52 | t.Fatalf("round %d: output got another feed", round) | 61 | t.Fatalf("round %d: output got another feed", round) | ||
| 53 | } | 62 | } | ||
| 54 | <-events | 63 | <-events | ||
| @@ -57,6 +66,11 @@ func TestOutputSetAttachesAndDetachesContents(t *testing.T) { | |||||
| 57 | t.Fatal(err) | 66 | t.Fatal(err) | ||
| 58 | } | 67 | } | ||
| 59 | } | 68 | } | ||
| 69 | + | for round := range detached { | |||
| 70 | + | if detached[round] != ports.ContentReturn(rets[round]) { | |||
| 71 | + | t.Fatalf("round %d: detached another feed than the one attached", round) | |||
| 72 | + | } | |||
| 73 | + | } | |||
| 60 | want := []string{"attach", "detach", "attach", "detach"} | 74 | want := []string{"attach", "detach", "attach", "detach"} | ||
| 61 | if len(order) != len(want) { | 75 | if len(order) != len(want) { | ||
| 62 | t.Fatalf("order %v, want %v", order, want) | 76 | t.Fatalf("order %v, want %v", order, want) | ||
| @@ -75,8 +89,9 @@ func TestOutputSetWaitDetachesContents(t *testing.T) { | |||||
| 75 | set := newOutputSet(ctx, nil) | 89 | set := newOutputSet(ctx, nil) | ||
| 76 | attacher := newMockcontentAttacher(t) | 90 | attacher := newMockcontentAttacher(t) | ||
| 77 | set.attach = attacher | 91 | set.attach = attacher | ||
| 78 | - | attacher.EXPECT().AttachOutput("A").Return(nil, nil).Once() | 92 | + | ret := portsmocks.NewMockContentReturn(t) |
| 79 | - | attacher.EXPECT().DetachOutput("A").Return().Once() | 93 | + | attacher.EXPECT().AttachOutput("A").Return(nil, ret).Once() |
| 94 | + | attacher.EXPECT().DetachOutput(ports.ContentReturn(ret)).Return().Once() | |||
| 80 | set.start(ctx, "A", func(ctx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, _ ports.OutputInstance) error { | 95 | set.start(ctx, "A", func(ctx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, _ ports.OutputInstance) error { | ||
| 81 | <-ctx.Done() | 96 | <-ctx.Done() | ||
| 82 | return nil | 97 | return nil | ||
M internal/app/outputs.go +14 -9
| @@ -20,12 +20,13 @@ type outputRun func(ctx context.Context, scenes <-chan ports.Scene, contents <-c | |||
| 20 | 20 | // content of each window, and takes the batches back through the returned | |
| 21 | 21 | // ports.ContentReturn. AttachOutput may block on the display goroutine: call it | |
| 22 | 22 | // from outside that goroutine. It returns a nil channel when wayland is gone. | |
| 23 | - | // Feeds are keyed by output name: attaching a name replaces its feed, and | |
| 24 | - | // DetachOutput stops whichever feed holds the name. Detach an output only once | |
| 25 | - | // it stopped, and before an output of the same name attaches again. | |
| 23 | + | // The returned ports.ContentReturn is also the handle of the feed: attaching | |
| 24 | + | // a name that already has a feed replaces it, and DetachOutput(handle) stops | |
| 25 | + | // that feed only, so a stale handle never stops a newer feed of the same | |
| 26 | + | // name. Detach an output once it stopped. | |
| 26 | 27 | type contentAttacher interface { | |
| 27 | 28 | AttachOutput(name string) (<-chan *ports.ContentBatch, ports.ContentReturn) | |
| 28 | - | DetachOutput(name string) | |
| 29 | + | DetachOutput(feed ports.ContentReturn) | |
| 29 | 30 | } | |
| 30 | 31 | ||
| 31 | 32 | // outputSet owns the running outputs and feeds them. Core sends one scene | |
| @@ -71,6 +72,9 @@ type runningOutput struct { | |||
| 71 | 72 | stop context.CancelFunc | |
| 72 | 73 | err error | |
| 73 | 74 | done chan struct{} | |
| 75 | + | // feed is the content feed AttachOutput returned (nil: none), the handle | |
| 76 | + | // to detach once the output stopped. | |
| 77 | + | feed ports.ContentReturn | |
| 74 | 78 | } | |
| 75 | 79 | ||
| 76 | 80 | func newOutputSet(ctx context.Context, captured chan<- ports.CaptureDone) *outputSet { | |
| @@ -106,6 +110,7 @@ func (s *outputSet) start(ctx context.Context, name string, run outputRun, wire | |||
| 106 | 110 | if s.attach != nil { | |
| 107 | 111 | contents, returns = s.attach.AttachOutput(name) | |
| 108 | 112 | } | |
| 113 | + | r.feed = returns | |
| 109 | 114 | s.outs[name] = r | |
| 110 | 115 | go func() { | |
| 111 | 116 | r.err = run(octx, r.scenes, contents, returns, r.cursor, r.captures, r.security, r.instance) | |
| @@ -249,15 +254,15 @@ func (s *outputSet) finish(name string) error { | |||
| 249 | 254 | <-r.done | |
| 250 | 255 | r.stop() | |
| 251 | 256 | s.failQueued(r) | |
| 252 | - | s.detach(name) | |
| 257 | + | s.detach(r) | |
| 253 | 258 | delete(s.outs, name) | |
| 254 | 259 | return r.err | |
| 255 | 260 | } | |
| 256 | 261 | ||
| 257 | 262 | // detach stops the content feed of an output that has stopped. | |
| 258 | - | func (s *outputSet) detach(name string) { | |
| 259 | - | if s.attach != nil { | |
| 260 | - | s.attach.DetachOutput(name) | |
| 263 | + | func (s *outputSet) detach(r *runningOutput) { | |
| 264 | + | if s.attach != nil && r.feed != nil { | |
| 265 | + | s.attach.DetachOutput(r.feed) | |
| 261 | 266 | } | |
| 262 | 267 | } | |
| 263 | 268 | ||
| @@ -275,7 +280,7 @@ func (s *outputSet) wait() error { | |||
| 275 | 280 | r.stop() | |
| 276 | 281 | <-r.done | |
| 277 | 282 | s.failQueued(r) | |
| 278 | - | s.detach(name) | |
| 283 | + | s.detach(r) | |
| 279 | 284 | if first == nil && r.err != nil { | |
| 280 | 285 | first = r.err | |
| 281 | 286 | } | |
| @@ -20,12 +20,13 @@ type outputRun func(ctx context.Context, scenes <-chan ports.Scene, contents <-c | |||||
| 20 | // content of each window, and takes the batches back through the returned | 20 | // content of each window, and takes the batches back through the returned | ||
| 21 | // ports.ContentReturn. AttachOutput may block on the display goroutine: call it | 21 | // ports.ContentReturn. AttachOutput may block on the display goroutine: call it | ||
| 22 | // from outside that goroutine. It returns a nil channel when wayland is gone. | 22 | // from outside that goroutine. It returns a nil channel when wayland is gone. | ||
| 23 | - | // Feeds are keyed by output name: attaching a name replaces its feed, and | 23 | + | // The returned ports.ContentReturn is also the handle of the feed: attaching |
| 24 | - | // DetachOutput stops whichever feed holds the name. Detach an output only once | 24 | + | // a name that already has a feed replaces it, and DetachOutput(handle) stops |
| 25 | - | // it stopped, and before an output of the same name attaches again. | 25 | + | // that feed only, so a stale handle never stops a newer feed of the same |
| 26 | + | // name. Detach an output once it stopped. | |||
| 26 | type contentAttacher interface { | 27 | type contentAttacher interface { | ||
| 27 | AttachOutput(name string) (<-chan *ports.ContentBatch, ports.ContentReturn) | 28 | AttachOutput(name string) (<-chan *ports.ContentBatch, ports.ContentReturn) | ||
| 28 | - | DetachOutput(name string) | 29 | + | DetachOutput(feed ports.ContentReturn) |
| 29 | } | 30 | } | ||
| 30 | 31 | ||||
| 31 | // outputSet owns the running outputs and feeds them. Core sends one scene | 32 | // outputSet owns the running outputs and feeds them. Core sends one scene | ||
| @@ -71,6 +72,9 @@ type runningOutput struct { | |||||
| 71 | stop context.CancelFunc | 72 | stop context.CancelFunc | ||
| 72 | err error | 73 | err error | ||
| 73 | done chan struct{} | 74 | done chan struct{} | ||
| 75 | + | // feed is the content feed AttachOutput returned (nil: none), the handle | |||
| 76 | + | // to detach once the output stopped. | |||
| 77 | + | feed ports.ContentReturn | |||
| 74 | } | 78 | } | ||
| 75 | 79 | ||||
| 76 | func newOutputSet(ctx context.Context, captured chan<- ports.CaptureDone) *outputSet { | 80 | func newOutputSet(ctx context.Context, captured chan<- ports.CaptureDone) *outputSet { | ||
| @@ -106,6 +110,7 @@ func (s *outputSet) start(ctx context.Context, name string, run outputRun, wire | |||||
| 106 | if s.attach != nil { | 110 | if s.attach != nil { | ||
| 107 | contents, returns = s.attach.AttachOutput(name) | 111 | contents, returns = s.attach.AttachOutput(name) | ||
| 108 | } | 112 | } | ||
| 113 | + | r.feed = returns | |||
| 109 | s.outs[name] = r | 114 | s.outs[name] = r | ||
| 110 | go func() { | 115 | go func() { | ||
| 111 | r.err = run(octx, r.scenes, contents, returns, r.cursor, r.captures, r.security, r.instance) | 116 | r.err = run(octx, r.scenes, contents, returns, r.cursor, r.captures, r.security, r.instance) | ||
| @@ -249,15 +254,15 @@ func (s *outputSet) finish(name string) error { | |||||
| 249 | <-r.done | 254 | <-r.done | ||
| 250 | r.stop() | 255 | r.stop() | ||
| 251 | s.failQueued(r) | 256 | s.failQueued(r) | ||
| 252 | - | s.detach(name) | 257 | + | s.detach(r) |
| 253 | delete(s.outs, name) | 258 | delete(s.outs, name) | ||
| 254 | return r.err | 259 | return r.err | ||
| 255 | } | 260 | } | ||
| 256 | 261 | ||||
| 257 | // detach stops the content feed of an output that has stopped. | 262 | // detach stops the content feed of an output that has stopped. | ||
| 258 | - | func (s *outputSet) detach(name string) { | 263 | + | func (s *outputSet) detach(r *runningOutput) { |
| 259 | - | if s.attach != nil { | 264 | + | if s.attach != nil && r.feed != nil { |
| 260 | - | s.attach.DetachOutput(name) | 265 | + | s.attach.DetachOutput(r.feed) |
| 261 | } | 266 | } | ||
| 262 | } | 267 | } | ||
| 263 | 268 | ||||
| @@ -275,7 +280,7 @@ func (s *outputSet) wait() error { | |||||
| 275 | r.stop() | 280 | r.stop() | ||
| 276 | <-r.done | 281 | <-r.done | ||
| 277 | s.failQueued(r) | 282 | s.failQueued(r) | ||
| 278 | - | s.detach(name) | 283 | + | s.detach(r) |
| 279 | if first == nil && r.err != nil { | 284 | if first == nil && r.err != nil { | ||
| 280 | first = r.err | 285 | first = r.err | ||
| 281 | } | 286 | } | ||