Merge pull request #313 from streamplace/eli/fix-stuck-playback-channel-2

director: handle segments that come in after session closes

pkg/director/director.go

···
 					segmentChan: make(chan struct{}),
 					op:          d.op,
 					packets:     make([]bus.PacketizedSegment, 0),
+					started:     make(chan struct{}),
 				}
 				d.streamSessions[not.Segment.RepoDID] = ss
 				g.Go(func() error {

pkg/director/stream_session.go

···
 	lastStatus     time.Time
 	lastStatusLock sync.Mutex
 	g              *errgroup.Group
+	started        chan struct{}
+	ctx            context.Context
 	packets        []bus.PacketizedSegment
 }
 
···
 	ss.g, ctx = errgroup.WithContext(ctx)
 	sid := livepeer.RandomTrailer(8)
 	ctx = log.WithLogValues(ctx, "sid", sid)
+	ss.ctx = ctx
 	log.Log(ctx, "starting stream session")
 	defer cancel()
 	spseg, err := not.Segment.ToStreamplaceSegment()
···
 	// 	})
 	// }
 
+	close(ss.started)
+
 	for {
 		select {
 		case <-ss.segmentChan:
···
 // non-fatal; if you actually want to melt the universe on an error you
 // should panic()
 func (ss *StreamSession) Go(ctx context.Context, f func() error) {
+	<-ss.started
 	ss.g.Go(func() error {
 		err := f()
 		if err != nil {
···
 }
 
 func (ss *StreamSession) NewSegment(ctx context.Context, not *media.NewSegmentNotification) error {
-	if ctx.Err() != nil {
-		return nil
-	}
-	ss.segmentChan <- struct{}{}
+	<-ss.started
+	go func() {
+		select {
+		case <-ss.ctx.Done():
+			return
+		case ss.segmentChan <- struct{}{}:
+		}
+	}()
 	aqt := aqtime.FromTime(not.Segment.StartTime)
 	ctx = log.WithLogValues(ctx, "segID", not.Segment.ID, "repoDID", not.Segment.RepoDID, "timestamp", aqt.FileSafeString())
 	err := ss.mod.CreateSegment(not.Segment)