webrtc_playback: buffer one segment before playback

pkg/media/packetize.go

···
 
 			videoOutput = append(videoOutput, samples)
 
+			// clockTime := buffer.Duration()
+			// dur := clockTime.AsDuration()
+			// if dur != nil {
+			// 	log.Log(ctx, "video duration", "duration", *dur)
+			// } else {
+			// 	log.Error(ctx, "no video duration", "samples", len(samples))
+			// }
+
 			return gst.FlowOK
 		},
 		EOSFunc: func(sink *app.Sink) {
···
 			if dur != nil {
 				segDur += *dur
 			} else {
-				return gst.FlowOK
+				log.Log(ctx, "no audio duration", "samples", len(samples))
+				err := fmt.Errorf("no audio duration")
+				pipeline.Error(err.Error(), err)
+				return gst.FlowError
 			}
 
 			return gst.FlowOK

pkg/media/packetize_test.go

···
 
 	"github.com/stretchr/testify/require"
 	"go.uber.org/goleak"
+	"golang.org/x/sync/errgroup"
 	"stream.place/streamplace/pkg/gstinit"
 	"stream.place/streamplace/pkg/media/segchanman"
 )
···
 	defer checkGStreamerLeaks(t, before)
 	ignore := goleak.IgnoreCurrent()
 	defer goleak.VerifyNone(t, ignore)
+
+	g, _ := errgroup.WithContext(context.Background())
+	for range streamplaceTestCount {
+		g.Go(func() error {
+			innerTestPacketize(t)
+			return nil
+		})
+	}
+	err := g.Wait()
+	require.NoError(t, err)
+}
+
+func innerTestPacketize(t *testing.T) {
 	filename := getFixture("sample-segment.mp4")
 	inputFile, err := os.Open(filename)
 	require.NoError(t, err)

pkg/media/webrtc_playback2.go

···
 	"github.com/google/uuid"
 	"github.com/pion/webrtc/v4"
 	"github.com/pion/webrtc/v4/pkg/media"
+	"golang.org/x/sync/errgroup"
 	"stream.place/streamplace/pkg/log"
 	"stream.place/streamplace/pkg/spmetrics"
 )
···
 				}
 			}()
 
+			lastPacketTime := time.Now()
+
+			p1 := <-packetQueue
+			p2 := <-packetQueue
+			bufPacketQueue := make(chan *PacketizedSegment, 1024)
+			go func() {
+				bufPacketQueue <- p1
+				bufPacketQueue <- p2
+				for {
+					select {
+					case <-ctx.Done():
+						return
+					case packet := <-packetQueue:
+						bufPacketQueue <- packet
+					}
+				}
+			}()
+
 			for {
 				select {
 				case <-ctx.Done():
 					return
-				case packet := <-packetQueue:
+				case packet := <-bufPacketQueue:
 					videoDur := packet.Duration / time.Duration(len(packet.Video))
 					audioDur := packet.Duration / time.Duration(len(packet.Audio))
-					go func() {
+					g, _ := errgroup.WithContext(ctx)
+
+					g.Go(func() error {
 						for _, video := range packet.Video {
+							// log.Log(ctx, "writing video sample", "duration", videoDur)
 							err := videoTrack.WriteSample(media.Sample{Data: video, Duration: videoDur})
 							if err != nil {
-								log.Error(ctx, "failed to write video sample", "error", err)
-								cancel()
-								return
+								return fmt.Errorf("failed to write video sample: %w", err)
 							}
 
 							select {
 							case <-ctx.Done():
-								return
+								return nil
 							case <-time.After(videoDur):
 								continue
 							}
 						}
-					}()
-					go func() {
+						return nil
+					})
+					g.Go(func() error {
+						log.Log(ctx, "time since last packet", "time", time.Since(lastPacketTime))
 						for _, audio := range packet.Audio {
 							err := audioTrack.WriteSample(media.Sample{Data: audio, Duration: audioDur})
 							if err != nil {
-								log.Error(ctx, "failed to write audio sample", "error", err)
-								cancel()
-								return
+								return fmt.Errorf("failed to write audio sample: %w", err)
 							}
 							select {
 							case <-ctx.Done():
-								return
+								return nil
 							case <-time.After(audioDur):
 								continue
 							}
 						}
-					}()
+						lastPacketTime = time.Now()
+						return nil
+					})
+
+					if err := g.Wait(); err != nil {
+						log.Error(ctx, "failed to write samples", "error", err)
+						cancel()
+					}
 				}
 			}
 		}()