Merge pull request #319 from streamplace/eli/new-webrtc-fixes

media: fix some new webrtc caching

pkg/bus/segchanman.go

···
 	}
 	curBuf = append(curBuf, seg)
 	if len(curBuf) > bufSize {
+		log.Warn(ctx, "segment buffer is too large, dropping oldest segment", "user", user, "rendition", rendition, "bufSize", bufSize, "len", len(curBuf))
 		curBuf = curBuf[1:]
 	}
 	b.segBuf[key] = curBuf
···
 		}(ch)
 	}
 }
+
+func (b *Bus) EndSession(ctx context.Context, user string, rendition string) {
+	b.segChansMutex.Lock()
+	defer b.segChansMutex.Unlock()
+	b.segBufMutex.Lock()
+	defer b.segBufMutex.Unlock()
+
+	key := segChanKey(user, rendition)
+	delete(b.segBuf, key)
+}

pkg/director/stream_session.go

···
 		// case <-time.After(time.Minute * 1):
 		case <-time.After(time.Second * 60):
 			log.Log(ctx, "no new segments for 1 minute, shutting down")
+			for _, r := range allRenditions {
+				ss.bus.EndSession(ctx, spseg.Creator, r.Name)
+			}
 			cancel()
 		}
 	}

pkg/media/packetize.go

···
 				return gst.FlowError
 			}
 
-			samples := buffer.Map(gst.MapRead).Bytes()
-			defer buffer.Unmap()
+			samples := buffer.Bytes()
 
 			videoOutput = append(videoOutput, samples)
 
···
 		},
 		EOSFunc: func(sink *app.Sink) {
 			log.Debug(ctx, "videoappsink EOSFunc")
-			// go func() {
-			// 	eosCh <- struct{}{}
-			// }()
 		},
 	})
 
···
 		NewSampleFunc: func(sink *app.Sink) gst.FlowReturn {
 			sample := sink.PullSample()
 			if sample == nil {
+				log.Warn(ctx, "audioappsink NewSampleFunc EOS")
 				return gst.FlowEOS
 			}
 
···
 				return gst.FlowError
 			}
 
-			samples := buffer.Map(gst.MapRead).Bytes()
-			defer buffer.Unmap()
+			samples := buffer.Bytes()
 
 			audioOutput = append(audioOutput, samples)
 
···
 			if dur != nil {
 				segDur += *dur
 			} else {
-				log.Log(ctx, "no audio duration", "samples", len(samples))
-				err := fmt.Errorf("no audio duration")
-				pipeline.Error(err.Error(), err)
+				log.Error(ctx, "no audio duration", "samples", len(samples))
 				return gst.FlowError
 			}
 
···
 		},
 		EOSFunc: func(sink *app.Sink) {
 			log.Debug(ctx, "audioappsink EOSFunc")
-			// go func() {
-			// 	eosCh <- struct{}{}
-			// }()
 		},
 	})
 
···
 		if err != nil {
 			log.Error(ctx, "failed to set pipeline to null state", "error", err)
 		}
+		err = pipeline.Remove(demuxBin.Element)
+		if err != nil {
+			log.Error(ctx, "failed to remove demux bin from bin", "error", err)
+		}
 	}()
-
-	// <-eosCh
-	// <-eosCh
 
 	err = <-busErr
 	if err != nil {

pkg/media/webrtc_ingest.go

···
 	// Create channel that is blocked until ICE Gathering is complete
 	gatherComplete := rtcrec.GatheringCompletePromise(peerConnection)
 
-	go func() {
-		ticker := time.NewTicker(time.Second * 1)
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case <-ticker.C:
-				state := pipeline.GetCurrentState()
-				log.Debug(ctx, "pipeline state", "state", state)
-			}
-		}
-	}()
 	// Setup complete! Now we boot up streaming in the background while returning the SDP offer to the user.
 
 	go func() {
 		ctx, cancel := context.WithCancel(ctx)
 		defer cancel()
 		defer func() { close(done) }()
+
+		go func() {
+			ticker := time.NewTicker(time.Second * 1)
+			for {
+				select {
+				case <-ctx.Done():
+					return
+				case <-ticker.C:
+					state := pipeline.GetCurrentState()
+					log.Debug(ctx, "pipeline state", "state", state)
+				}
+			}
+		}()
 
 		go func() {
 			if err := HandleBusMessages(ctx, pipeline); err != nil {
···
 		if err := videoSrcElem.SetState(gst.StateNull); err != nil {
 			log.Log(ctx, "failed to set videoSrcElem state to null", "error", err)
 		}
+
+		log.Log(ctx, "webrtc ingest pipeline done")
 
 	}()
 	select {

pkg/media/webrtc_playback2.go

···
 		ctx, cancel := context.WithCancel(ctx)
 		defer cancel()
 
-		started := time.Now()
-		elapsed := time.Duration(0)
 		latency := time.Duration(0)
-
-		go func() {
-			for {
-				select {
-				case <-ctx.Done():
-					return
-				case <-time.After(time.Second * 5):
-					log.Log(ctx, "check elapsed", "elapsed", elapsed, "duration", time.Since(started))
-				}
-			}
-		}()
 
 		packetQueue := make(chan *bus.PacketizedSegment, 1024)
 		go func() {