stream.place
Live video on the AT Protocol
director: handle segments that come in after session closes
+15
-4
+1
pkg/director/director.go
+1
pkg/director/director.go
+14
-4
pkg/director/stream_session.go
+14
-4
pkg/director/stream_session.go
···
39
39
lastStatus time.Time
40
40
lastStatusLock sync.Mutex
41
41
g *errgroup.Group
42
+
started chan struct{}
43
+
ctx context.Context
42
44
packets []bus.PacketizedSegment
43
45
}
44
46
···
47
49
ss.g, ctx = errgroup.WithContext(ctx)
48
50
sid := livepeer.RandomTrailer(8)
49
51
ctx = log.WithLogValues(ctx, "sid", sid)
52
+
ss.ctx = ctx
50
53
log.Log(ctx, "starting stream session")
51
54
defer cancel()
52
55
spseg, err := not.Segment.ToStreamplaceSegment()
···
94
97
// })
95
98
// }
96
99
100
+
close(ss.started)
101
+
97
102
for {
98
103
select {
99
104
case <-ss.segmentChan:
···
112
117
// non-fatal; if you actually want to melt the universe on an error you
113
118
// should panic()
114
119
func (ss *StreamSession) Go(ctx context.Context, f func() error) {
120
+
<-ss.started
115
121
ss.g.Go(func() error {
116
122
err := f()
117
123
if err != nil {
···
122
128
}
123
129
124
130
func (ss *StreamSession) NewSegment(ctx context.Context, not *media.NewSegmentNotification) error {
125
-
if ctx.Err() != nil {
126
-
return nil
127
-
}
128
-
ss.segmentChan <- struct{}{}
131
+
<-ss.started
132
+
go func() {
133
+
select {
134
+
case <-ss.ctx.Done():
135
+
return
136
+
case ss.segmentChan <- struct{}{}:
137
+
}
138
+
}()
129
139
aqt := aqtime.FromTime(not.Segment.StartTime)
130
140
ctx = log.WithLogValues(ctx, "segID", not.Segment.ID, "repoDID", not.Segment.RepoDID, "timestamp", aqt.FileSafeString())
131
141
err := ss.mod.CreateSegment(not.Segment)