improvements to gstreamer memory handling (#83)

* gstreamer: move SetState(null) to end of functions

* media: split out webrtc into two files

* .gitignore: oom

* media: polish to avoid memory leaks

* media: do more memory cleanup

* media: implement proper gstreamer leak testing

* thumbnail: add queue to remove leak

* thumbnail: polish up leak tester

* media: move leak test to its own file

* leaks: fix debug string

* concat: well it's not always 6 but it's better!

* gstreamer: misc memory improvements

* test: fix gstreamer leaks count

* gstreamer: try without .clear()

* build: statically link libgstcoretracers.a

* build: flip off mediaparser leak detector

* build: make test on linux

Changelog: feature

.github/workflows/build.yaml

···
       - name: make node
         run: |
           sudo apt install -y podman
-          make in-container IN_CONTAINER_CMD="make"
+          make in-container IN_CONTAINER_CMD="make && make test && make check"
 
   linux-arm64:
     name: linux-arm64

.gitignore

···
 *.log
 *.heap
 /api
+oom

Makefile

···
 		-D "gst-plugins-ugly:gpl=enabled" \
 		-D "x264:asm=enabled" \
 		-D "gstreamer-full:gst-full=enabled" \
-		-D "gstreamer-full:gst-full-plugins=libgstopusparse.a;libgstcodectimestamper.a;libgstrtp.a;libgstaudioresample.a;libgstlibav.a;libgstmatroska.a;libgstmultifile.a;libgstjpeg.a;libgstaudiotestsrc.a;libgstaudioconvert.a;libgstaudioparsers.a;libgstfdkaac.a;libgstisomp4.a;libgstapp.a;libgstvideoconvertscale.a;libgstvideobox.a;libgstvideorate.a;libgstpng.a;libgstcompositor.a;libgstaudiorate.a;libgstx264.a;libgstopus.a;libgstvideotestsrc.a;libgstvideoparsersbad.a;libgstaudioparsers.a;libgstmpegtsmux.a;libgstmpegtsdemux.a;libgstplayback.a;libgsttypefindfunctions.a" \
+		-D "gstreamer-full:gst-full-plugins=libgstopusparse.a;libgstcodectimestamper.a;libgstrtp.a;libgstaudioresample.a;libgstlibav.a;libgstmatroska.a;libgstmultifile.a;libgstjpeg.a;libgstaudiotestsrc.a;libgstaudioconvert.a;libgstaudioparsers.a;libgstfdkaac.a;libgstisomp4.a;libgstapp.a;libgstvideoconvertscale.a;libgstvideobox.a;libgstvideorate.a;libgstpng.a;libgstcompositor.a;libgstaudiorate.a;libgstx264.a;libgstopus.a;libgstvideotestsrc.a;libgstvideoparsersbad.a;libgstaudioparsers.a;libgstmpegtsmux.a;libgstmpegtsdemux.a;libgstplayback.a;libgsttypefindfunctions.a;libgstcoretracers.a" \
 		-D "gstreamer-full:gst-full-libraries=gstreamer-controller-1.0,gstreamer-plugins-base-1.0,gstreamer-pbutils-1.0" \
 		-D "gstreamer-full:gst-full-elements=coreelements:concat,filesrc,filesink,queue,queue2,multiqueue,typefind,tee,capsfilter,fakesink,identity" \
 		-D "gstreamer-full:bad=enabled" \
···
 		-D "gstreamer-full:gst-full-typefind-functions=" \
 		-D "gstreamer-full:glib_assert=false" \
 		-D "gstreamer:glib_assert=false" \
+		-D "gstreamer:coretracers=enabled" \
 		-D "gst-plugins-good:glib_assert=false" \
 		-D "gst-plugins-bad:glib_assert=false" \
 		-D "gst-plugins-base:glib_assert=false" \

go.mod

···
 	github.com/Microsoft/go-winio v0.6.2 // indirect
 	github.com/ProtonMail/go-crypto v1.0.0 // indirect
 	github.com/RussellLuo/slidingwindow v0.0.0-20200528002341-535bb99d338b // indirect
+	github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d // indirect
 	github.com/agnivade/levenshtein v1.2.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bits-and-blooms/bitset v1.10.0 // indirect

go.sum

···
 github.com/StackExchange/wmi v1.2.1/go.mod h1:rcmrprowKIVzvc+NUiLncP2uuArMWLCbu9SBzvHz7e8=
 github.com/VictoriaMetrics/fastcache v1.12.2 h1:N0y9ASrJ0F6h0QaC3o6uJb3NIZ9VKLjCM7NQbSmF7WI=
 github.com/VictoriaMetrics/fastcache v1.12.2/go.mod h1:AmC+Nzz1+3G2eCPapF6UcsnkThDcMsQicp4xDukwJYI=
+github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d h1:licZJFw2RwpHMqeKTCYkitsPqHNxTmd4SNR5r94FGM8=
+github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d/go.mod h1:asat636LX7Bqt5lYEZ27JNDcqxfjdBQuJ/MM4CN/Lzo=
 github.com/agnivade/levenshtein v1.2.0 h1:U9L4IOT0Y3i0TIlUIDJ7rVUziKi/zPbrJGaFrtYH3SY=
 github.com/agnivade/levenshtein v1.2.0/go.mod h1:QVVI16kDrtSuwcpd0p1+xMC6Z/VfhtCyDIjcwga4/DU=
 github.com/alexbrainman/goissue34681 v0.0.0-20191006012335-3fc7a47baff5 h1:iW0a5ljuFxkLGPNem5Ui+KBjFJzKg4Fv2fnxe4dvzpM=

pkg/cmd/streamplace.go

···
 	"strconv"
 	"strings"
 	"syscall"
+	"time"
 
 	"golang.org/x/term"
 	"stream.place/streamplace/pkg/aqhttp"
···
 	fs.StringVar(&cli.RelayHost, "relay-host", "wss://bsky.network", "websocket url for relay firehose")
 	fs.Bool("insecure", false, "DEPRECATED, does nothing.")
 	fs.StringVar(&cli.Color, "color", "", "'true' to enable colorized logging, 'false' to disable")
+	fs.BoolVar(&cli.Thumbnail, "thumbnail", true, "enable thumbnail generation")
 
 	version := fs.Bool("version", false, "print version and exit")
 
···
 
 	if cli.WHIPTest != "" {
 		group.Go(func() error {
-			return WHIP(strings.Split(cli.WHIPTest, " "))
+			err := WHIP(strings.Split(cli.WHIPTest, " "))
+			log.Warn(ctx, "WHIP test complete, sleeping for 3 seconds and shutting down gstreamer")
+			time.Sleep(time.Second * 3)
+			// gst.Deinit()
+			log.Warn(ctx, "gst deinit complete, exiting")
+			return err
 		})
 	}
 

pkg/cmd/whip.go

···
 					}
 				}
 			})
+			go func() {
+				<-ctx.Done()
+				if conn.peerConnection != nil {
+					conn.peerConnection.Close()
+				}
+			}()
 			return nil
 		})
 	}
···
 		defer ticker.Stop()
 
 		for {
-			<-ticker.C
-			for i, duration := range accumulators {
-				trackType := "video"
-				if i == 1 {
-					trackType = "audio"
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				for i, duration := range accumulators {
+					trackType := "video"
+					if i == 1 {
+						trackType = "audio"
+					}
+					target := startTime.Add(time.Duration(accumulators[i]))
+					diff := time.Since(target)
+					log.Debug(ctx, "elapsed duration", "track", trackType, "duration", duration, "diff", diff)
 				}
-				target := startTime.Add(time.Duration(accumulators[i]))
-				diff := time.Since(target)
-				log.Debug(ctx, "elapsed duration", "track", trackType, "duration", duration, "diff", diff)
 			}
 		}
 	}()
···
 			return err
 		}
 	}
+
+	<-ctx.Done()
+	err = pipeline.BlockSetState(gst.StateNull)
+	if err != nil {
+		return err
+	}
+
 	select {
 	case err := <-errCh:
 		return err

pkg/config/config.go

···
 	Color                  string
 	LivepeerGatewayURL     string
 	WHIPTest               string
+	Thumbnail              bool
 
 	dataDirFlags []*string
 }

pkg/director/stream_session.go

···
 
 	ss.bus.Publish(spseg.Creator, spseg)
 
-	go func() {
-		err := ss.Thumbnail(ctx, spseg.Creator, not)
-		if err != nil {
-			log.Error(ctx, "could not create thumbnail", "error", err)
-		}
-	}()
+	if ss.cli.Thumbnail {
+		go func() {
+			err := ss.Thumbnail(ctx, spseg.Creator, not)
+			if err != nil {
+				log.Error(ctx, "could not create thumbnail", "error", err)
+			}
+		}()
+	}
 
 	if ss.cli.LivepeerGatewayURL != "" {
 		go func() {
···
 		return err
 	}
 	defer fd.Close()
-	err = ss.mm.Thumbnail(ctx, r, fd)
+	err = media.Thumbnail(ctx, r, fd)
 	if err != nil {
 		return err
 	}

pkg/media/concat.go

···
 		return nil, nil, fmt.Errorf("failed to get input queue audio src pad")
 	}
 
+	go func() {
+		<-ctx.Done()
+		inputQueue.SetState(gst.StateNull)
+		inputQueue = nil
+		inputQueuePadVideoSink = nil
+		inputQueuePadVideoSrc = nil
+		inputQueuePadAudioSink = nil
+		inputQueuePadAudioSrc = nil
+	}()
+
 	// streamsynchronizer
 	streamsynchronizer, err := gst.NewElementWithProperties("streamsynchronizer", map[string]any{})
 	if err != nil {
 		return nil, nil, fmt.Errorf("failed to create streamsynchronizer element: %w", err)
 	}
+	go func() {
+		<-ctx.Done()
+		streamsynchronizer.SetState(gst.StateNull)
+		streamsynchronizer = nil
+	}()
 	err = pipeline.Add(streamsynchronizer)
 	if err != nil {
 		return nil, nil, fmt.Errorf("failed to add streamsynchronizer to pipeline: %w", err)
···
 	}
 
 	// output multiqueue
-	outputQueue, err := gst.NewElementWithProperties("multiqueue", map[string]any{})
+	outputQueue, err := gst.NewElementWithProperties("multiqueue", map[string]any{
+		"name": "concat-output-queue",
+	})
 	if err != nil {
 		return nil, nil, fmt.Errorf("failed to create multiqueue element: %w", err)
 	}
···
 	if outputQueuePadAudioSink == nil {
 		return nil, nil, fmt.Errorf("failed to get output queue audio sink pad")
 	}
+	go func() {
+		<-ctx.Done()
+		outputQueue.SetState(gst.StateNull)
+		outputQueue = nil
+		outputQueuePadVideoSink = nil
+		outputQueuePadAudioSink = nil
+	}()
 
 	// linking
 
···
 				log.Debug(ctx, "got segment", "file", file.Filepath)
 				allFiles <- file.Data
 				if len(file.Data) == 0 {
-					log.Warn(ctx, "no more segments")
+					log.Warn(ctx, "no more segments, stopping segment reader")
 					return
 				}
 			}
 		}
 	}()
 
+	segCount := 0
+
 	// nextFile is the primary loop that pops off a file, creates new demuxer elements for it,
 	// and pushes into the pipeline
 	var nextFile func()
 	nextFile = func() {
+		mySegCount := segCount
+		segCount += 1
+		segDone := make(chan struct{})
+		log.Debug(ctx, "moving to next file", "segCount", mySegCount)
 		pr, pw := io.Pipe()
 		go func() {
 			select {
···
 				return
 			case bs := <-allFiles:
 				if len(bs) == 0 {
-					log.Warn(ctx, "no more segments")
+					log.Warn(ctx, "no more segments, ending stream")
+					pr.Close()
+					pw.Close()
 					cancel()
 					return
 				}
···
 			}
 		}()
 
-		demux, err := gst.NewElementWithProperties("qtdemux", map[string]any{})
+		demux, err := gst.NewElementWithProperties("qtdemux", map[string]any{
+			"name": fmt.Sprintf("concat-demux-%d", mySegCount),
+		})
 		if err != nil {
 			log.Error(ctx, "failed to create demux element", "error", err)
 			cancel()
···
 					if count == 0 {
 						// don't keep going if our context is done
 						if ctx.Err() == nil {
-							nextFile()
+							go nextFile()
+							segDone <- struct{}{}
 						}
+					} else {
+						log.Debug(ctx, "demux has more pads, waiting for them to close")
 					}
 					return gst.PadProbeRemove
 				})
···
 		}
 
 		appsrc, err := gst.NewElementWithProperties("appsrc", map[string]any{
+			"name":    fmt.Sprintf("concat-appsrc-%d", mySegCount),
 			"is-live": true,
 		})
 		if err != nil {
···
 							cancel()
 							return
 						}
-						log.Debug(ctx, "EOF, ending stream", "length", read)
+						log.Debug(ctx, "EOF, ending segment", "length", read)
 						done()
 						return
 					} else {
···
 				self.PushBuffer(buffer)
 
 				if uint(read) < length {
-					log.Debug(ctx, "short write, ending stream", "length", read)
+					log.Debug(ctx, "short write, ending segment", "length", read)
 					done()
 				}
 			},
···
 			cancel()
 			return
 		}
+
+		select {
+		case <-ctx.Done():
+		case <-segDone:
+		}
+
+		log.Debug(ctx, "ending segment")
+		demux.SetState(gst.StateNull)
+		src.SetCallbacks(&app.SourceCallbacks{})
+		appsrc.SetState(gst.StateNull)
+		pr.Close()
+		pw.Close()
 	}
 
 	// fire it up!

pkg/media/concat_test.go

···
+package media
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"math/rand"
+	"os"
+	"strconv"
+	"testing"
+
+	"github.com/go-gst/go-gst/gst"
+	"github.com/go-gst/go-gst/gst/app"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/goleak"
+	"stream.place/streamplace/pkg/gstinit"
+	"stream.place/streamplace/pkg/log"
+	"stream.place/streamplace/pkg/media/segchanman"
+)
+
+type TestConcatStreamer struct {
+	fileName string
+	data     []byte
+	count    int
+}
+
+func (t *TestConcatStreamer) SubscribeSegment(ctx context.Context, user string, rendition string) <-chan *segchanman.Seg {
+	if len(t.data) == 0 {
+		panic("test file empty")
+	}
+	ch := make(chan *segchanman.Seg)
+	go func() {
+		if t.count == 5 {
+			ch <- &segchanman.Seg{
+				Data:     nil,
+				Filepath: "",
+			}
+		} else {
+			fmt.Println("writing segment " + strconv.Itoa(t.count) + " with random number " + strconv.Itoa(rand.Intn(100)))
+			ch <- &segchanman.Seg{
+				Data:     t.data,
+				Filepath: t.fileName,
+			}
+			t.count += 1
+		}
+	}()
+	return ch
+}
+
+func (t *TestConcatStreamer) UnsubscribeSegment(ctx context.Context, user string, rendition string, ch <-chan *segchanman.Seg) {
+}
+
+func TestConcat(t *testing.T) {
+	gstinit.InitGST()
+	// before := getLeakCount(t)
+	// defer checkGStreamerLeaks(t, before)
+	ignore := goleak.IgnoreCurrent()
+	defer goleak.VerifyNone(t, ignore)
+
+	innnerTestConcat(t)
+	after := getLeakCount(t)
+	if after != 6 {
+		fmt.Println("leaks", after)
+	}
+}
+
+// This function remains in scope for the duration of a single users' playback
+func innnerTestConcat(t *testing.T) {
+
+	ctx := log.WithDebugValue(context.Background(), map[string]map[string]int{"func": {"ConcatStream": 9, "TestConcat": 9}})
+	ctx = log.WithLogValues(ctx, "func", "TestConcat")
+	ctx, cancel := context.WithCancel(ctx)
+	// defer cancel()
+
+	pipeline, err := gst.NewPipeline("TestConcat")
+	require.NoError(t, err)
+
+	go func() {
+		HandleBusMessages(ctx, pipeline)
+		cancel()
+	}()
+
+	filename := getFixture("sample-segment.mp4")
+	inputFile, err := os.Open(filename)
+	require.NoError(t, err)
+	bs, err := io.ReadAll(inputFile)
+	require.NoError(t, err)
+	tcs := &TestConcatStreamer{
+		fileName: getFixture("sample-segment.mp4"),
+		data:     bs,
+	}
+
+	outputQueue, done, err := ConcatStream(ctx, pipeline, "fakeuser", "fakerendition", tcs)
+	require.NoError(t, err)
+
+	go func() {
+		select {
+		case <-ctx.Done():
+			return
+		case <-done:
+			cancel()
+		}
+	}()
+
+	videoPad := outputQueue.GetStaticPad("src_0")
+	require.NotNil(t, videoPad)
+
+	audioPad := outputQueue.GetStaticPad("src_1")
+	require.NotNil(t, audioPad)
+
+	videoAppSink, err := gst.NewElementWithProperties("appsink", map[string]interface{}{
+		"name":        "videoappsink",
+		"sync":        false,
+		"wait-on-eos": false,
+	})
+	require.NoError(t, err)
+	err = pipeline.Add(videoAppSink)
+	require.NoError(t, err)
+
+	videoAppSinkPadSink := videoAppSink.GetStaticPad("sink")
+	require.NotNil(t, videoAppSinkPadSink)
+
+	audioAppSink, err := gst.NewElementWithProperties("appsink", map[string]interface{}{
+		"name":        "audioappsink",
+		"sync":        false,
+		"wait-on-eos": false,
+	})
+	require.NoError(t, err)
+	err = pipeline.Add(audioAppSink)
+	require.NoError(t, err)
+
+	audioAppSinkPadSink := audioAppSink.GetStaticPad("sink")
+	require.NotNil(t, audioAppSinkPadSink)
+
+	ok := videoPad.Link(videoAppSinkPadSink)
+	require.Equal(t, gst.PadLinkOK, ok)
+
+	ok = audioPad.Link(audioAppSinkPadSink)
+	require.Equal(t, gst.PadLinkOK, ok)
+
+	videoTotalBytes := 0
+	audioTotalBytes := 0
+
+	videoappsink := app.SinkFromElement(videoAppSink)
+	videoappsink.SetCallbacks(&app.SinkCallbacks{
+		NewSampleFunc: func(sink *app.Sink) gst.FlowReturn {
+			sample := sink.PullSample()
+			if sample == nil {
+				return gst.FlowEOS
+			}
+
+			buffer := sample.GetBuffer()
+			if buffer == nil {
+				return gst.FlowError
+			}
+
+			samples := buffer.Map(gst.MapRead).Bytes()
+			defer buffer.Unmap()
+
+			videoTotalBytes += len(samples)
+
+			return gst.FlowOK
+		},
+		EOSFunc: func(sink *app.Sink) {
+			log.Warn(ctx, "videoappsink EOSFunc")
+			cancel()
+		},
+	})
+
+	audioappsink := app.SinkFromElement(audioAppSink)
+	audioappsink.SetCallbacks(&app.SinkCallbacks{
+		NewSampleFunc: func(sink *app.Sink) gst.FlowReturn {
+			sample := sink.PullSample()
+			if sample == nil {
+				return gst.FlowEOS
+			}
+
+			buffer := sample.GetBuffer()
+			if buffer == nil {
+				return gst.FlowError
+			}
+
+			samples := buffer.Map(gst.MapRead).Bytes()
+			defer buffer.Unmap()
+
+			audioTotalBytes += len(samples)
+
+			return gst.FlowOK
+		},
+		EOSFunc: func(sink *app.Sink) {
+			log.Warn(ctx, "audioappsink EOSFunc")
+			cancel()
+		},
+	})
+
+	// Start the pipeline
+
+	err = pipeline.SetState(gst.StatePlaying)
+	require.NoError(t, err)
+
+	<-ctx.Done()
+
+	err = pipeline.BlockSetState(gst.StateNull)
+	require.NoError(t, err)
+	pipeline.Remove(videoAppSink)
+	pipeline.Remove(audioAppSink)
+	videoAppSink.SetState(gst.StateNull)
+	audioAppSink.SetState(gst.StateNull)
+	videoappsink.SetCallbacks(&app.SinkCallbacks{})
+	audioappsink.SetCallbacks(&app.SinkCallbacks{})
+	pipeline.Clear()
+
+	require.Greater(t, videoTotalBytes, 1000000)
+	require.Greater(t, audioTotalBytes, 40000)
+}

pkg/media/gstreamer.go

···
 
 		// Retrieve the buffer from the sample.
 		buffer := sample.GetBuffer()
+		bs := buffer.Map(gst.MapRead).Bytes()
+		defer buffer.Unmap()
 
-		_, err := io.Copy(output, buffer.Reader())
+		_, err := output.Write(bs)
 
 		if err != nil {
 			panic(err)
···
 
 // basic test to make sure gstreamer functionality is working
 func SelfTest(ctx context.Context) error {
+	ctx = log.WithLogValues(ctx, "mediafunc", "SelfTest")
 	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
 	defer cancel()
 	f, err := test.Files.Open("fixtures/sample-segment.mp4")
 	if err != nil {
-		return err
+		return fmt.Errorf("failed to open test file: %w", err)
 	}
 	defer f.Close()
 	bs, err := io.ReadAll(f)
 	if err != nil {
-		return err
+		return fmt.Errorf("failed to read test file: %w", err)
 	}
 
-	pipeline, err := gst.NewPipelineFromString("appsrc name=src ! appsink name=sink")
+	pipeline, err := gst.NewPipeline("self-test")
 	if err != nil {
-		return err
+		return fmt.Errorf("failed to create pipeline: %w", err)
 	}
 
-	srcele, err := pipeline.GetElementByName("src")
+	srcele, err := gst.NewElementWithProperties("appsrc", map[string]interface{}{
+		"name": "self-test-src",
+	})
 	if err != nil {
-		return err
+		return fmt.Errorf("failed to create appsrc element: %w", err)
 	}
-	if srcele == nil {
-		return fmt.Errorf("srcele not found")
+	err = pipeline.Add(srcele)
+	if err != nil {
+		return fmt.Errorf("failed to add appsrc to pipeline: %w", err)
+	}
+
+	sinkele, err := gst.NewElementWithProperties("appsink", map[string]interface{}{
+		"name": "self-test-sink",
+	})
+	if err != nil {
+		return fmt.Errorf("failed to create appsink element: %w", err)
+	}
+	err = pipeline.Add(sinkele)
+	if err != nil {
+		return fmt.Errorf("failed to add appsink to pipeline: %w", err)
+	}
+
+	err = srcele.Link(sinkele)
+	if err != nil {
+		return fmt.Errorf("failed to link appsrc to appsink: %w", err)
 	}
+
+	// pipeline, err := gst.NewPipelineFromString("appsrc name=src ! appsink name=sink")
+	// if err != nil {
+	// 	return err
+	// }
+
 	src := app.SrcFromElement(srcele)
 	src.SetCallbacks(&app.SourceCallbacks{
 		NeedDataFunc: func(self *app.Source, _ uint) {
···
 			buffer.Map(gst.MapWrite).WriteData(bs)
 			defer buffer.Unmap()
 			self.PushBuffer(buffer)
+			log.Debug(ctx, "ending stream")
 			self.EndStream()
 		},
 	})
 
-	mainLoop := glib.NewMainLoop(glib.MainContextDefault(), false)
-
 	output := &bytes.Buffer{}
-	sinkele, err := pipeline.GetElementByName("sink")
+
 	if err != nil {
-		return err
+		return fmt.Errorf("unexpected error: %w", err)
 	}
-	if sinkele == nil {
-		return fmt.Errorf("sinkele not found")
-	}
+
 	appsink := app.SinkFromElement(sinkele)
 	appsink.SetCallbacks(&app.SinkCallbacks{
 		NewSampleFunc: func(sink *app.Sink) gst.FlowReturn {
···
 			return gst.FlowOK
 		},
 		EOSFunc: func(sink *app.Sink) {
+			log.Debug(ctx, "EOSFunc")
 			cancel()
 		},
 	})
 
-	// Start the pipeline
-	pipeline.SetState(gst.StatePlaying)
-
 	go func() {
-		<-ctx.Done()
-		mainLoop.Quit()
+		HandleBusMessages(ctx, pipeline)
+		cancel()
 	}()
 
-	mainLoop.Run()
+	// Start the pipeline
+	log.Debug(ctx, "setting pipeline to playing state")
+	err = pipeline.SetState(gst.StatePlaying)
+	if err != nil {
+		return fmt.Errorf("failed to set pipeline to playing state: %w", err)
+	}
+
+	<-ctx.Done()
 
 	if len(output.Bytes()) < 1 {
 		return fmt.Errorf("got a zero-byte buffer from SelfTest")
 	}
+
+	err = pipeline.BlockSetState(gst.StateNull)
+	if err != nil {
+		return fmt.Errorf("failed to set pipeline to null state: %w", err)
+	}
+
 	return nil
 }
 

pkg/media/leak_test.go

···
+package media
+
+import (
+	"bufio"
+	"fmt"
+	"os"
+	"path/filepath"
+	"regexp"
+	"runtime"
+	"runtime/debug"
+	"strings"
+	"sync"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/acarl005/stripansi"
+	"github.com/stretchr/testify/require"
+)
+
+const GST_DEBUG_NEEDED = "leaks:9,GST_TRACER:9"
+const LEAK_LINE = "GST_TRACER :0:: object-alive"
+
+var LEAK_DONE_REGEX = regexp.MustCompile(`listed\s+(\d+)\s+alive\s+objects`)
+
+var LeakReport = []string{}
+var LeakReportMutex sync.Mutex
+var LeakDoneCh = make(chan struct{})
+
+func TestMain(m *testing.M) {
+	gstDebug := os.Getenv("GST_DEBUG")
+	if gstDebug == "" {
+		gstDebug = GST_DEBUG_NEEDED
+	} else {
+		gstDebug = fmt.Sprintf("%s,%s", gstDebug, GST_DEBUG_NEEDED)
+	}
+	os.Setenv("GST_DEBUG", gstDebug)
+	os.Setenv("GST_TRACERS", "leaks")
+	os.Setenv("GST_LEAKS_TRACER_SIG", "1")
+	debug.SetGCPercent(5)
+
+	f, err := os.MkdirTemp("", "")
+	if err != nil {
+		panic(err)
+	}
+	fName := filepath.Join(f, "leak.log")
+	err = syscall.Mkfifo(fName, 0640)
+	if err != nil {
+		panic(err)
+	}
+	os.Setenv("GST_DEBUG_FILE", fName)
+
+	go func() {
+		pipe, err := os.OpenFile(fName, os.O_RDONLY, 0640)
+		if err != nil {
+			panic(err)
+		}
+		defer pipe.Close()
+		// Read and print each line from FD
+		scanner := bufio.NewScanner(pipe)
+		for scanner.Scan() {
+			line := scanner.Text()
+			fmt.Println(line)
+			line = stripansi.Strip(line)
+			if strings.Contains(line, LEAK_LINE) {
+				LeakReportMutex.Lock()
+				LeakReport = append(LeakReport, line)
+				LeakReportMutex.Unlock()
+			} else if LEAK_DONE_REGEX.MatchString(line) {
+				LeakDoneCh <- struct{}{}
+			} else {
+				continue
+			}
+		}
+		if err := scanner.Err(); err != nil {
+			panic(err)
+		}
+	}()
+	os.Exit(m.Run())
+}
+
+func getLeakCount(t *testing.T) int {
+	process, err := os.FindProcess(os.Getpid())
+	LeakReportMutex.Lock()
+	LeakReport = []string{}
+	LeakReportMutex.Unlock()
+
+	ch := make(chan struct{})
+	done := false
+
+	go func() {
+		thing := &[]byte{}
+		runtime.SetFinalizer(thing, func(thing *[]byte) {
+			done = true
+			ch <- struct{}{}
+		})
+	}()
+
+	go func() {
+		runtime.GC()
+		runtime.GC()
+		for {
+			if done {
+				break
+			}
+			runtime.GC()
+			runtime.GC()
+			time.Sleep(500 * time.Millisecond)
+		}
+	}()
+
+	<-ch
+	time.Sleep(1 * time.Second)
+
+	err = process.Signal(os.Signal(syscall.SIGUSR1))
+	require.NoError(t, err)
+
+	<-LeakDoneCh
+
+	LeakReportMutex.Lock()
+	after := len(LeakReport)
+	LeakReportMutex.Unlock()
+	return after
+}
+
+func checkGStreamerLeaks(t *testing.T, expected int) {
+	leaks := getLeakCount(t)
+	if leaks > expected {
+		LeakReportMutex.Lock()
+		for _, l := range LeakReport {
+			fmt.Println(l)
+		}
+		LeakReportMutex.Unlock()
+	}
+	require.Equal(t, expected, len(LeakReport), "Leaks found")
+}

pkg/media/media_data_parser.go

···
 	"stream.place/streamplace/pkg/model"
 )
 
-func (mm *MediaManager) ParseSegmentMediaData(ctx context.Context, mp4bs []byte) (*model.SegmentMediaData, error) {
+func ParseSegmentMediaData(ctx context.Context, mp4bs []byte) (*model.SegmentMediaData, error) {
 	ctx = log.WithLogValues(ctx, "GStreamerFunc", "ParseSegmentMediaData")
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	pipelineSlice := []string{
-		"appsrc name=appsrc ! qtdemux name=demux ! fakesink",
+		"appsrc name=appsrc ! qtdemux name=demux ! fakesink sync=false",
 	}
 
 	pipeline, err := gst.NewPipelineFromString(strings.Join(pipelineSlice, "\n"))

pkg/media/media_data_parser_test.go

···
+package media
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/goleak"
+	"stream.place/streamplace/pkg/gstinit"
+	"stream.place/streamplace/pkg/log"
+)
+
+func TestMediaDataParser(t *testing.T) {
+	gstinit.InitGST()
+	before := getLeakCount(t)
+	// defer checkGStreamerLeaks(t, before+1)
+	defer func() {
+		after := getLeakCount(t)
+		if after > before {
+			fmt.Printf("detected %d leaks", after-before)
+		}
+	}()
+	ignore := goleak.IgnoreCurrent()
+	defer goleak.VerifyNone(t, ignore)
+
+	// Open input file
+	inputFile, err := os.Open(getFixture("sample-segment.mp4"))
+	require.NoError(t, err)
+	defer inputFile.Close()
+	bs, err := io.ReadAll(inputFile)
+	require.NoError(t, err)
+
+	ctx := log.WithDebugValue(context.Background(), map[string]map[string]int{"GStreamerFunc": {"ParseSegmentMediaData": 9}})
+	mediaData, err := ParseSegmentMediaData(ctx, bs)
+	require.NoError(t, err)
+	require.NotNil(t, mediaData)
+	require.Greater(t, mediaData.Duration, int64(0), "Video duration should not be empty")
+}

pkg/media/segmenter.go

···
 		if appsink == nil {
 			panic("appsink should not be nil")
 		}
+
 		appsink.SetCallbacks(&app.SinkCallbacks{
 			NewSampleFunc: WriterNewSample(ctx, buf),
 			EOSFunc: func(sink *app.Sink) {

pkg/media/thumbnail.go

···
 	"stream.place/streamplace/pkg/log"
 )
 
-func (mm *MediaManager) Thumbnail(ctx context.Context, r io.Reader, w io.Writer) error {
+func Thumbnail(ctx context.Context, r io.Reader, w io.Writer) error {
 	ctx = log.WithLogValues(ctx, "function", "Thumbnail")
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
 	pipelineSlice := []string{
-		"appsrc name=appsrc ! qtdemux ! decodebin ! videoconvert ! videoscale ! video/x-raw,width=[1,720],height=[1,720],pixel-aspect-ratio=1/1 ! pngenc snapshot=true ! appsink name=appsink",
+		"appsrc name=appsrc ! qtdemux name=demux ! decodebin ! videoconvert ! videoscale ! capsfilter name=capsfilter caps=video/x-raw,width=[1,1280],height=[1,720],pixel-aspect-ratio=1/1 ! queue ! pngenc snapshot=true ! appsink name=appsink",
 	}
 
 	pipeline, err := gst.NewPipelineFromString(strings.Join(pipelineSlice, "\n"))

pkg/media/thumbnail_test.go

···
+package media
+
+import (
+	"bytes"
+	"context"
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/goleak"
+	"stream.place/streamplace/pkg/gstinit"
+	"stream.place/streamplace/pkg/log"
+)
+
+func TestThumbnail(t *testing.T) {
+	gstinit.InitGST()
+	before := getLeakCount(t)
+	defer checkGStreamerLeaks(t, before+1)
+	ignore := goleak.IgnoreCurrent()
+	defer goleak.VerifyNone(t, ignore)
+
+	// Open input file
+	inputFile, err := os.Open(getFixture("sample-segment.mp4"))
+	require.NoError(t, err)
+	defer inputFile.Close()
+
+	thumbnail := bytes.Buffer{}
+	ctx := log.WithDebugValue(context.Background(), map[string]map[string]int{"function": {"Thumbnail": 9}})
+	err = Thumbnail(ctx, inputFile, &thumbnail)
+	require.NoError(t, err)
+	require.NotNil(t, thumbnail)
+	require.Greater(t, thumbnail.Len(), 0, "Thumbnail buffer should not be empty")
+}

pkg/media/validate.go

···
 	if err != nil {
 		return err
 	}
-	mediaData, err := mm.ParseSegmentMediaData(ctx, buf)
+	mediaData, err := ParseSegmentMediaData(ctx, buf)
 	if err != nil {
 		return err
 	}

pkg/media/webrtc.go

···
-package media
-
-import (
-	"context"
-	"fmt"
-	"strings"
-	"time"
-
-	"github.com/go-gst/go-gst/gst"
-	"github.com/go-gst/go-gst/gst/app"
-	"github.com/google/uuid"
-	"github.com/pion/rtcp"
-	"github.com/pion/webrtc/v4"
-	"github.com/pion/webrtc/v4/pkg/media"
-	"stream.place/streamplace/pkg/log"
-	"stream.place/streamplace/pkg/spmetrics"
-)
-
-// we have a bug that prevents us from correctly probing video durations
-// a lot of the time. so when we don't have them we use the last duration
-// that we had, and when we don't have that we use a default duration
-var DEFAULT_DURATION = time.Duration(32 * time.Millisecond)
-
-// This function remains in scope for the duration of a single users' playback
-func (mm *MediaManager) WebRTCPlayback(ctx context.Context, user string, rendition string, offer *webrtc.SessionDescription) (*webrtc.SessionDescription, error) {
-	uu, err := uuid.NewV7()
-	if err != nil {
-		return nil, err
-	}
-	ctx = log.WithLogValues(ctx, "webrtcID", uu.String())
-	ctx, cancel := context.WithCancel(ctx)
-
-	ctx = log.WithLogValues(ctx, "mediafunc", "WebRTCPlayback")
-
-	pipelineSlice := []string{
-		"h264parse name=videoparse ! video/x-h264,stream-format=byte-stream ! appsink name=videoappsink",
-		"opusparse name=audioparse ! appsink name=audioappsink",
-	}
-
-	pipeline, err := gst.NewPipelineFromString(strings.Join(pipelineSlice, "\n"))
-	if err != nil {
-		return nil, fmt.Errorf("failed to create GStreamer pipeline: %w", err)
-	}
-
-	go func() {
-		HandleBusMessages(ctx, pipeline)
-		cancel()
-	}()
-
-	outputQueue, done, err := ConcatStream(ctx, pipeline, user, rendition, mm)
-	if err != nil {
-		return nil, fmt.Errorf("failed to get output queue: %w", err)
-	}
-	go func() {
-		select {
-		case <-ctx.Done():
-			return
-		case <-done:
-			cancel()
-		}
-	}()
-	// queuePadVideo := outputQueue.GetRequestPad("src_%u")
-	// if queuePadVideo == nil {
-	// 	return nil, fmt.Errorf("failed to get queue video pad")
-	// }
-	// queuePadAudio := outputQueue.GetRequestPad("src_%u")
-	// if queuePadAudio == nil {
-	// 	return nil, fmt.Errorf("failed to get queue audio pad")
-	// }
-
-	videoParse, err := pipeline.GetElementByName("videoparse")
-	if err != nil {
-		return nil, fmt.Errorf("failed to get video sink element from pipeline: %w", err)
-	}
-	err = outputQueue.Link(videoParse)
-	if err != nil {
-		return nil, fmt.Errorf("failed to link output queue to video parse: %w", err)
-	}
-
-	audioParse, err := pipeline.GetElementByName("audioparse")
-	if err != nil {
-		return nil, fmt.Errorf("failed to get audio parse element from pipeline: %w", err)
-	}
-	err = outputQueue.Link(audioParse)
-	if err != nil {
-		return nil, fmt.Errorf("failed to link output queue to audio parse: %w", err)
-	}
-
-	videoappsinkele, err := pipeline.GetElementByName("videoappsink")
-	if err != nil {
-		return nil, fmt.Errorf("failed to get video sink element from pipeline: %w", err)
-	}
-
-	audioappsinkele, err := pipeline.GetElementByName("audioappsink")
-	if err != nil {
-		return nil, fmt.Errorf("failed to get audio sink element from pipeline: %w", err)
-	}
-
-	// Create a new RTCPeerConnection
-	peerConnection, err := mm.webrtcAPI.NewPeerConnection(mm.webrtcConfig)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create WebRTC peer connection: %w", err)
-	}
-	go func() {
-		<-ctx.Done()
-		if cErr := peerConnection.Close(); cErr != nil {
-			log.Log(ctx, "cannot close peerConnection: %v\n", cErr)
-		}
-	}()
-
-	videoTrack, err := webrtc.NewTrackLocalStaticSample(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264}, "video", "pion")
-	if err != nil {
-		return nil, fmt.Errorf("failed to create video track: %w", err)
-	}
-	videoRTPSender, err := peerConnection.AddTrack(videoTrack)
-	if err != nil {
-		return nil, fmt.Errorf("failed to add video track to peer connection: %w", err)
-	}
-
-	audioTrack, err := webrtc.NewTrackLocalStaticSample(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus}, "audio", "pion")
-	if err != nil {
-		return nil, fmt.Errorf("failed to create audio track: %w", err)
-	}
-	audioRTPSender, err := peerConnection.AddTrack(audioTrack)
-	if err != nil {
-		return nil, fmt.Errorf("failed to add audio track to peer connection: %w", err)
-	}
-
-	// Set the remote SessionDescription
-	if err = peerConnection.SetRemoteDescription(*offer); err != nil {
-		return nil, fmt.Errorf("failed to set remote description: %w", err)
-	}
-
-	// Create answer
-	answer, err := peerConnection.CreateAnswer(nil)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create answer: %w", err)
-	}
-
-	// Sets the LocalDescription, and starts our UDP listeners
-	if err = peerConnection.SetLocalDescription(answer); err != nil {
-		return nil, fmt.Errorf("failed to set local description: %w", err)
-	}
-
-	// Create channel that is blocked until ICE Gathering is complete
-	gatherComplete := webrtc.GatheringCompletePromise(peerConnection)
-
-	// Setup complete! Now we boot up streaming in the background while returning the SDP offer to the user.
-
-	go func() {
-		<-ctx.Done()
-		pipeline.BlockSetState(gst.StateNull)
-	}()
-
-	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)
-			}
-		}
-	}()
-
-	var lastVideoDuration = &DEFAULT_DURATION
-
-	go func() {
-
-		videoappsink := app.SinkFromElement(videoappsinkele)
-		videoappsink.SetCallbacks(&app.SinkCallbacks{
-			NewSampleFunc: func(sink *app.Sink) gst.FlowReturn {
-				sample := sink.PullSample()
-				if sample == nil {
-					return gst.FlowEOS
-				}
-
-				buffer := sample.GetBuffer()
-				if buffer == nil {
-					return gst.FlowError
-				}
-
-				samples := buffer.Map(gst.MapRead).Bytes()
-				defer buffer.Unmap()
-				clockTime := buffer.Duration()
-				dur := clockTime.AsDuration()
-				mediaSample := media.Sample{Data: samples}
-				if dur != nil {
-					mediaSample.Duration = *dur
-					lastVideoDuration = dur
-				} else if lastVideoDuration != nil {
-					mediaSample.Duration = *lastVideoDuration
-				} else {
-					log.Log(ctx, "no video duration", "samples", len(samples))
-					// cancel()
-					return gst.FlowOK
-				}
-
-				if err := videoTrack.WriteSample(mediaSample); err != nil {
-					log.Log(ctx, "failed to write video sample", "error", err)
-					cancel()
-				}
-
-				return gst.FlowOK
-			},
-			EOSFunc: func(sink *app.Sink) {
-				log.Warn(ctx, "videoappsink EOSFunc")
-				cancel()
-			},
-		})
-
-		audioappsink := app.SinkFromElement(audioappsinkele)
-		audioappsink.SetCallbacks(&app.SinkCallbacks{
-			NewSampleFunc: func(sink *app.Sink) gst.FlowReturn {
-				sample := sink.PullSample()
-				if sample == nil {
-					return gst.FlowEOS
-				}
-
-				buffer := sample.GetBuffer()
-				if buffer == nil {
-					return gst.FlowError
-				}
-
-				samples := buffer.Map(gst.MapRead).Bytes()
-				defer buffer.Unmap()
-
-				clockTime := buffer.Duration()
-				dur := clockTime.AsDuration()
-				mediaSample := media.Sample{Data: samples}
-				if dur != nil {
-					mediaSample.Duration = *dur
-				} else {
-					log.Log(ctx, "no audio duration", "samples", len(samples))
-					// cancel()
-					return gst.FlowOK
-				}
-				if err := audioTrack.WriteSample(mediaSample); err != nil {
-					log.Log(ctx, "failed to write audio sample", "error", err)
-					return gst.FlowOK
-				}
-
-				return gst.FlowOK
-			},
-			EOSFunc: func(sink *app.Sink) {
-				log.Warn(ctx, "audioappsink EOSFunc")
-				cancel()
-			},
-		})
-
-		// Start the pipeline
-		pipeline.SetState(gst.StatePlaying)
-		spmetrics.ViewerInc(user)
-		defer spmetrics.ViewerDec(user)
-
-		go func() {
-			rtcpBuf := make([]byte, 1500)
-			for {
-				if _, _, rtcpErr := videoRTPSender.Read(rtcpBuf); rtcpErr != nil {
-					return
-				}
-			}
-		}()
-
-		go func() {
-			rtcpBuf := make([]byte, 1500)
-			for {
-				if _, _, rtcpErr := audioRTPSender.Read(rtcpBuf); rtcpErr != nil {
-					return
-				}
-			}
-		}()
-
-		// Set the handler for ICE connection state
-		// This will notify you when the peer has connected/disconnected
-		peerConnection.OnICEConnectionStateChange(func(connectionState webrtc.ICEConnectionState) {
-			log.Log(ctx, "Connection State has changed", "state", connectionState.String())
-			if connectionState == webrtc.ICEConnectionStateConnected {
-				// iceConnectedCtxCancel()
-			}
-		})
-
-		// Set the handler for Peer connection state
-		// This will notify you when the peer has connected/disconnected
-		peerConnection.OnConnectionStateChange(func(s webrtc.PeerConnectionState) {
-			log.Log(ctx, "Peer Connection State has changed", "state", s.String())
-
-			if s == webrtc.PeerConnectionStateFailed || s == webrtc.PeerConnectionStateClosed {
-				// Wait until PeerConnection has had no network activity for 30 seconds or another failure. It may be reconnected using an ICE Restart.
-				// Use webrtc.PeerConnectionStateDisconnected if you are interested in detecting faster timeout.
-				// Note that the PeerConnection may come back from PeerConnectionStateDisconnected.
-				log.Log(ctx, "Peer Connection has gone to failed, exiting")
-				cancel()
-			}
-		})
-
-		<-ctx.Done()
-	}()
-	select {
-	case <-gatherComplete:
-		return peerConnection.LocalDescription(), nil
-	case <-ctx.Done():
-		return nil, ctx.Err()
-	}
-}
-
-// This function remains in scope for the duration of a single users' playback
-func (mm *MediaManager) WebRTCIngest(ctx context.Context, offer *webrtc.SessionDescription, signer MediaSigner) (*webrtc.SessionDescription, error) {
-	uu, err := uuid.NewV7()
-	if err != nil {
-		return nil, err
-	}
-
-	ctx, cancel := context.WithCancel(ctx)
-	ctx = log.WithLogValues(ctx, "webrtcID", uu.String(), "mediafunc", "WebRTCIngest")
-
-	// Setup the codecs you want to use.
-	// We'll use a VP8 and Opus but you can also define your own
-
-	// Create a new RTCPeerConnection
-	peerConnection, err := mm.webrtcAPI.NewPeerConnection(mm.webrtcConfig)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create WebRTC peer connection: %w", err)
-	}
-
-	// Allow us to receive 1 audio track, and 1 video track
-	if _, err = peerConnection.AddTransceiverFromKind(webrtc.RTPCodecTypeAudio); err != nil {
-		return nil, fmt.Errorf("failed to add audio transceiver: %w", err)
-	} else if _, err = peerConnection.AddTransceiverFromKind(webrtc.RTPCodecTypeVideo); err != nil {
-		return nil, fmt.Errorf("failed to add video transceiver: %w", err)
-	}
-
-	pipelineSlice := []string{
-		"multiqueue name=queue",
-		"appsrc format=time is-live=true do-timestamp=true name=videosrc ! capsfilter caps=application/x-rtp ! rtph264depay ! capsfilter caps=video/x-h264,stream-format=byte-stream,alignment=nal ! h264parse ! h264timestamper ! identity ! queue.sink_0",
-		"appsrc format=time is-live=true do-timestamp=true name=audiosrc ! capsfilter caps=application/x-rtp,media=audio,encoding-name=OPUS,payload=111 ! rtpopusdepay ! opusdec use-inband-fec=true ! audiorate ! opusenc ! queue.sink_1",
-	}
-
-	pipeline, err := gst.NewPipelineFromString(strings.Join(pipelineSlice, "\n"))
-	if err != nil {
-		return nil, fmt.Errorf("failed to create GStreamer pipeline: %w", err)
-	}
-
-	go func() {
-		HandleBusMessages(ctx, pipeline)
-		cancel()
-	}()
-
-	queue, err := pipeline.GetElementByName("queue")
-	if err != nil {
-		return nil, fmt.Errorf("failed to get queue element from pipeline: %w", err)
-	}
-
-	signerElem, err := mm.SegmentAndSignElem(ctx, signer)
-	if err != nil {
-		return nil, fmt.Errorf("failed create signer element: %w", err)
-	}
-	err = pipeline.Add(signerElem)
-	if err != nil {
-		return nil, fmt.Errorf("failed to add signer element to pipeline: %w", err)
-	}
-
-	// err = queue.Link(signerElem)
-	// if err != nil {
-	// 	return nil, fmt.Errorf("failed to link queue to signer element: %w", err)
-	// }
-	videoSrcPads, err := queue.GetSrcPads()
-	if err != nil {
-		return nil, fmt.Errorf("failed to get videoSrcPads from queue: %w", err)
-	}
-	if len(videoSrcPads) != 2 {
-		return nil, fmt.Errorf("failed to get videoSrcPads from queue")
-	}
-	videoSrcPad := videoSrcPads[0]
-	audioSrcPad := videoSrcPads[1]
-
-	signerElemPads, err := signerElem.GetPads()
-	if err != nil {
-		return nil, fmt.Errorf("failed to get signerElemPads from signer element: %w", err)
-	}
-	if len(signerElemPads) != 2 {
-		return nil, fmt.Errorf("failed to get signerElemPads from signer element")
-	}
-	signerElemVideoPad := signerElemPads[0]
-	signerElemAudioPad := signerElemPads[1]
-	videoSrcPad.Link(signerElemVideoPad)
-	audioSrcPad.Link(signerElemAudioPad)
-
-	videoSrcElem, err := pipeline.GetElementByName("videosrc")
-	if err != nil {
-		return nil, fmt.Errorf("failed to get videoSrcElem element from pipeline: %w", err)
-	}
-	videoSrc := app.SrcFromElement(videoSrcElem)
-
-	audioSrcElem, err := pipeline.GetElementByName("audiosrc")
-	if err != nil {
-		return nil, fmt.Errorf("failed to get audioSrcElem element from pipeline: %w", err)
-	}
-	audioSrc := app.SrcFromElement(audioSrcElem)
-
-	go func() {
-		<-ctx.Done()
-		pipeline.BlockSetState(gst.StateNull)
-	}()
-
-	go func() {
-		<-ctx.Done()
-		if cErr := peerConnection.Close(); cErr != nil {
-			log.Log(ctx, "cannot close peerConnection: %v\n", cErr)
-		}
-	}()
-
-	// Set the remote SessionDescription
-	if err = peerConnection.SetRemoteDescription(*offer); err != nil {
-		return nil, fmt.Errorf("failed to set remote description: %w", err)
-	}
-
-	// Create answer
-	answer, err := peerConnection.CreateAnswer(nil)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create answer: %w", err)
-	}
-
-	// Sets the LocalDescription, and starts our UDP listeners
-	if err = peerConnection.SetLocalDescription(answer); err != nil {
-		return nil, fmt.Errorf("failed to set local description: %w", err)
-	}
-
-	// Create channel that is blocked until ICE Gathering is complete
-	gatherComplete := webrtc.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() {
-		log.Debug(ctx, "starting pipeline")
-
-		// Start the pipeline
-		err = pipeline.SetState(gst.StatePlaying)
-		if err != nil {
-			log.Log(ctx, "failed to set pipeline state", "error", err)
-			cancel()
-		}
-
-		// Set the handler for ICE connection state
-		// This will notify you when the peer has connected/disconnected
-		peerConnection.OnICEConnectionStateChange(func(connectionState webrtc.ICEConnectionState) {
-			log.Log(ctx, "Connection State has changed", "state", connectionState.String())
-			if connectionState == webrtc.ICEConnectionStateConnected {
-				// iceConnectedCtxCancel()
-			}
-		})
-
-		// Set the handler for Peer connection state
-		// This will notify you when the peer has connected/disconnected
-		peerConnection.OnConnectionStateChange(func(s webrtc.PeerConnectionState) {
-			log.Log(ctx, "Peer Connection State has changed", "state", s.String())
-
-			if s == webrtc.PeerConnectionStateFailed || s == webrtc.PeerConnectionStateDisconnected {
-				// Wait until PeerConnection has had no network activity for 30 seconds or another failure. It may be reconnected using an ICE Restart.
-				// Use webrtc.PeerConnectionStateDisconnected if you are interested in detecting faster timeout.
-				// Note that the PeerConnection may come back from PeerConnectionStateDisconnected.
-				log.Log(ctx, "Peer Connection has ended, exiting", "state", s.String())
-				cancel()
-			}
-		})
-
-		log.Warn(ctx, "setting OnTrack")
-		peerConnection.OnTrack(func(track *webrtc.TrackRemote, _ *webrtc.RTPReceiver) {
-			log.Warn(ctx, "OnTrack")
-			if track.Kind() == webrtc.RTPCodecTypeVideo {
-				// Send a PLI on an interval so that the publisher is pushing a keyframe every rtcpPLIInterval
-				go func() {
-					ticker := time.NewTicker(time.Second * 5)
-					for {
-						select {
-						case <-ctx.Done():
-							return
-						case <-ticker.C:
-							rtcpSendErr := peerConnection.WriteRTCP([]rtcp.Packet{&rtcp.PictureLossIndication{MediaSSRC: uint32(track.SSRC())}})
-							if rtcpSendErr != nil {
-								log.Log(ctx, "failed to send rtcp packet", "error", rtcpSendErr)
-								cancel()
-								return
-							}
-						}
-					}
-				}()
-
-				codecName := strings.Split(track.Codec().RTPCodecCapability.MimeType, "/")[1]
-				log.Log(ctx, "Track has started", "payloadType", track.PayloadType(), "codecName", codecName)
-
-				// appSrc := pipelineForCodec(track, codecName)
-				buf := make([]byte, 1400)
-				for {
-					i, _, readErr := track.Read(buf)
-					if readErr != nil {
-						log.Log(ctx, "failed to read track", "error", readErr)
-						cancel()
-						return
-					}
-					log.Debug(ctx, "read video track", "bytes", i)
-
-					ret := videoSrc.PushBuffer(gst.NewBufferFromBytes(buf[:i]))
-					if ret != gst.FlowOK {
-						log.Log(ctx, "failed to push buffer", "error", ret)
-						cancel()
-						return
-					}
-					// state := pipeline.GetCurrentState()
-					// if state != gst.StatePlaying {
-					// 	log.Warn(ctx, "pipeline state is not playing, consider running with GST_DEBUG=*:5 to find out why", "state", state)
-					// 	cancel()
-					// 	return
-					// }
-				}
-			}
-			if track.Kind() == webrtc.RTPCodecTypeAudio {
-
-				codecName := strings.Split(track.Codec().RTPCodecCapability.MimeType, "/")[1]
-				log.Log(ctx, "Track has started", "payloadType", track.PayloadType(), "codecName", codecName)
-
-				buf := make([]byte, 1400)
-				for {
-					i, _, readErr := track.Read(buf)
-					if readErr != nil {
-						log.Log(ctx, "failed to read track", "error", readErr)
-						cancel()
-						return
-					}
-					// log.Log(ctx, "read audio track", "bytes", i)
-
-					ret := audioSrc.PushBuffer(gst.NewBufferFromBytes(buf[:i]))
-					if ret != gst.FlowOK {
-						log.Log(ctx, "failed to push buffer", "error", ret)
-						cancel()
-						return
-					}
-					// state := pipeline.GetCurrentState()
-					// if state != gst.StatePlaying {
-					// 	log.Warn(ctx, "pipeline state is not playing, consider running with GST_DEBUG=*:5 to find out why", "state", state)
-					// 	cancel()
-					// 	return
-					// }
-				}
-			}
-		})
-
-		<-ctx.Done()
-	}()
-	select {
-	case <-gatherComplete:
-		return peerConnection.LocalDescription(), nil
-	case <-ctx.Done():
-		return nil, ctx.Err()
-	}
-}

pkg/media/webrtc_ingest.go

···
+package media
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/go-gst/go-gst/gst"
+	"github.com/go-gst/go-gst/gst/app"
+	"github.com/google/uuid"
+	"github.com/pion/rtcp"
+	"github.com/pion/webrtc/v4"
+	"stream.place/streamplace/pkg/log"
+)
+
+// This function remains in scope for the duration of a single users' playback
+func (mm *MediaManager) WebRTCIngest(ctx context.Context, offer *webrtc.SessionDescription, signer MediaSigner) (*webrtc.SessionDescription, error) {
+	uu, err := uuid.NewV7()
+	if err != nil {
+		return nil, err
+	}
+
+	ctx, cancel := context.WithCancel(ctx)
+	ctx = log.WithLogValues(ctx, "webrtcID", uu.String(), "mediafunc", "WebRTCIngest")
+
+	// Setup the codecs you want to use.
+	// We'll use a VP8 and Opus but you can also define your own
+
+	// Create a new RTCPeerConnection
+	peerConnection, err := mm.webrtcAPI.NewPeerConnection(mm.webrtcConfig)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create WebRTC peer connection: %w", err)
+	}
+
+	// Allow us to receive 1 audio track, and 1 video track
+	if _, err = peerConnection.AddTransceiverFromKind(webrtc.RTPCodecTypeAudio); err != nil {
+		return nil, fmt.Errorf("failed to add audio transceiver: %w", err)
+	} else if _, err = peerConnection.AddTransceiverFromKind(webrtc.RTPCodecTypeVideo); err != nil {
+		return nil, fmt.Errorf("failed to add video transceiver: %w", err)
+	}
+
+	pipelineSlice := []string{
+		"multiqueue name=queue",
+		"appsrc format=time is-live=true do-timestamp=true name=videosrc ! capsfilter caps=application/x-rtp ! rtph264depay ! capsfilter caps=video/x-h264,stream-format=byte-stream,alignment=nal ! h264parse ! h264timestamper ! identity ! queue.sink_0",
+		"appsrc format=time is-live=true do-timestamp=true name=audiosrc ! capsfilter caps=application/x-rtp,media=audio,encoding-name=OPUS,payload=111 ! rtpopusdepay ! opusdec use-inband-fec=true ! audiorate ! opusenc ! queue.sink_1",
+	}
+
+	pipeline, err := gst.NewPipelineFromString(strings.Join(pipelineSlice, "\n"))
+	if err != nil {
+		return nil, fmt.Errorf("failed to create GStreamer pipeline: %w", err)
+	}
+
+	go func() {
+		HandleBusMessages(ctx, pipeline)
+		cancel()
+	}()
+
+	queue, err := pipeline.GetElementByName("queue")
+	if err != nil {
+		return nil, fmt.Errorf("failed to get queue element from pipeline: %w", err)
+	}
+
+	signerElem, err := mm.SegmentAndSignElem(ctx, signer)
+	if err != nil {
+		return nil, fmt.Errorf("failed create signer element: %w", err)
+	}
+	err = pipeline.Add(signerElem)
+	if err != nil {
+		return nil, fmt.Errorf("failed to add signer element to pipeline: %w", err)
+	}
+
+	// err = queue.Link(signerElem)
+	// if err != nil {
+	// 	return nil, fmt.Errorf("failed to link queue to signer element: %w", err)
+	// }
+	videoSrcPads, err := queue.GetSrcPads()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get videoSrcPads from queue: %w", err)
+	}
+	if len(videoSrcPads) != 2 {
+		return nil, fmt.Errorf("failed to get videoSrcPads from queue")
+	}
+	videoSrcPad := videoSrcPads[0]
+	audioSrcPad := videoSrcPads[1]
+
+	signerElemPads, err := signerElem.GetPads()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get signerElemPads from signer element: %w", err)
+	}
+	if len(signerElemPads) != 2 {
+		return nil, fmt.Errorf("failed to get signerElemPads from signer element")
+	}
+	signerElemVideoPad := signerElemPads[0]
+	signerElemAudioPad := signerElemPads[1]
+	videoSrcPad.Link(signerElemVideoPad)
+	audioSrcPad.Link(signerElemAudioPad)
+
+	videoSrcElem, err := pipeline.GetElementByName("videosrc")
+	if err != nil {
+		return nil, fmt.Errorf("failed to get videoSrcElem element from pipeline: %w", err)
+	}
+	videoSrc := app.SrcFromElement(videoSrcElem)
+
+	audioSrcElem, err := pipeline.GetElementByName("audiosrc")
+	if err != nil {
+		return nil, fmt.Errorf("failed to get audioSrcElem element from pipeline: %w", err)
+	}
+	audioSrc := app.SrcFromElement(audioSrcElem)
+
+	go func() {
+		<-ctx.Done()
+		if cErr := peerConnection.Close(); cErr != nil {
+			log.Log(ctx, "cannot close peerConnection: %v\n", cErr)
+		}
+	}()
+
+	// Set the remote SessionDescription
+	if err = peerConnection.SetRemoteDescription(*offer); err != nil {
+		return nil, fmt.Errorf("failed to set remote description: %w", err)
+	}
+
+	// Create answer
+	answer, err := peerConnection.CreateAnswer(nil)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create answer: %w", err)
+	}
+
+	// Sets the LocalDescription, and starts our UDP listeners
+	if err = peerConnection.SetLocalDescription(answer); err != nil {
+		return nil, fmt.Errorf("failed to set local description: %w", err)
+	}
+
+	// Create channel that is blocked until ICE Gathering is complete
+	gatherComplete := webrtc.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() {
+		log.Debug(ctx, "starting pipeline")
+
+		// Start the pipeline
+		err = pipeline.SetState(gst.StatePlaying)
+		if err != nil {
+			log.Log(ctx, "failed to set pipeline state", "error", err)
+			cancel()
+		}
+
+		// Set the handler for ICE connection state
+		// This will notify you when the peer has connected/disconnected
+		peerConnection.OnICEConnectionStateChange(func(connectionState webrtc.ICEConnectionState) {
+			log.Log(ctx, "Connection State has changed", "state", connectionState.String())
+			if connectionState == webrtc.ICEConnectionStateConnected {
+				// iceConnectedCtxCancel()
+			}
+		})
+
+		// Set the handler for Peer connection state
+		// This will notify you when the peer has connected/disconnected
+		peerConnection.OnConnectionStateChange(func(s webrtc.PeerConnectionState) {
+			log.Log(ctx, "Peer Connection State has changed", "state", s.String())
+
+			if s == webrtc.PeerConnectionStateFailed || s == webrtc.PeerConnectionStateDisconnected {
+				// Wait until PeerConnection has had no network activity for 30 seconds or another failure. It may be reconnected using an ICE Restart.
+				// Use webrtc.PeerConnectionStateDisconnected if you are interested in detecting faster timeout.
+				// Note that the PeerConnection may come back from PeerConnectionStateDisconnected.
+				log.Log(ctx, "Peer Connection has ended, exiting", "state", s.String())
+				cancel()
+			}
+		})
+
+		videoFirst := false
+		audioFirst := false
+
+		log.Warn(ctx, "setting OnTrack")
+		peerConnection.OnTrack(func(track *webrtc.TrackRemote, _ *webrtc.RTPReceiver) {
+			log.Warn(ctx, "OnTrack")
+			if track.Kind() == webrtc.RTPCodecTypeVideo {
+				// Send a PLI on an interval so that the publisher is pushing a keyframe every rtcpPLIInterval
+				go func() {
+					ticker := time.NewTicker(time.Second * 5)
+					for {
+						select {
+						case <-ctx.Done():
+							return
+						case <-ticker.C:
+							rtcpSendErr := peerConnection.WriteRTCP([]rtcp.Packet{&rtcp.PictureLossIndication{MediaSSRC: uint32(track.SSRC())}})
+							if rtcpSendErr != nil {
+								log.Log(ctx, "failed to send rtcp packet", "error", rtcpSendErr)
+								cancel()
+								return
+							}
+						}
+					}
+				}()
+
+				codecName := strings.Split(track.Codec().RTPCodecCapability.MimeType, "/")[1]
+				log.Log(ctx, "Track has started", "payloadType", track.PayloadType(), "codecName", codecName)
+
+				// appSrc := pipelineForCodec(track, codecName)
+				buf := make([]byte, 1400)
+				for {
+					i, _, readErr := track.Read(buf)
+					if readErr != nil {
+						log.Log(ctx, "failed to read track", "error", readErr)
+						cancel()
+						return
+					}
+					if ctx.Err() != nil {
+						return
+					}
+					if !videoFirst {
+						videoFirst = true
+						log.Debug(ctx, "got video data", "len", len(buf[:i]))
+					}
+
+					gbuf := gst.NewBufferWithSize(int64(len(buf[:i])))
+					gbuf.Map(gst.MapWrite).WriteData(buf[:i])
+					defer gbuf.Unmap()
+
+					ret := videoSrc.PushBuffer(gbuf)
+					if ret != gst.FlowOK {
+						log.Log(ctx, "failed to push buffer", "error", ret)
+						cancel()
+						return
+					}
+					// state := pipeline.GetCurrentState()
+					// if state != gst.StatePlaying {
+					// 	log.Warn(ctx, "pipeline state is not playing, consider running with GST_DEBUG=*:5 to find out why", "state", state)
+					// 	cancel()
+					// 	return
+					// }
+				}
+			}
+			if track.Kind() == webrtc.RTPCodecTypeAudio {
+
+				codecName := strings.Split(track.Codec().RTPCodecCapability.MimeType, "/")[1]
+				log.Log(ctx, "Track has started", "payloadType", track.PayloadType(), "codecName", codecName)
+
+				buf := make([]byte, 1400)
+				for {
+					i, _, readErr := track.Read(buf)
+					if readErr != nil {
+						log.Log(ctx, "failed to read track", "error", readErr)
+						cancel()
+						return
+					}
+					if ctx.Err() != nil {
+						return
+					}
+					if !audioFirst {
+						audioFirst = true
+						log.Debug(ctx, "got audio data", "len", len(buf[:i]))
+					}
+
+					gbuf := gst.NewBufferWithSize(int64(len(buf[:i])))
+					gbuf.Map(gst.MapWrite).WriteData(buf[:i])
+					defer gbuf.Unmap()
+					ret := audioSrc.PushBuffer(gbuf)
+					if ret != gst.FlowOK {
+						log.Log(ctx, "failed to push buffer", "error", ret)
+						cancel()
+						return
+					}
+					// state := pipeline.GetCurrentState()
+					// if state != gst.StatePlaying {
+					// 	log.Warn(ctx, "pipeline state is not playing, consider running with GST_DEBUG=*:5 to find out why", "state", state)
+					// 	cancel()
+					// 	return
+					// }
+				}
+			}
+		})
+
+		<-ctx.Done()
+
+		err = pipeline.BlockSetState(gst.StateNull)
+		if err != nil {
+			log.Log(ctx, "failed to set pipeline state to null", "error", err)
+		}
+
+		audioSrcElem.SetState(gst.StateNull)
+		if err != nil {
+			log.Log(ctx, "failed to set audioSrcElem state to null", "error", err)
+		}
+
+		videoSrcElem.SetState(gst.StateNull)
+		if err != nil {
+			log.Log(ctx, "failed to set videoSrcElem state to null", "error", err)
+		}
+
+	}()
+	select {
+	case <-gatherComplete:
+		return peerConnection.LocalDescription(), nil
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}

pkg/media/webrtc_playback.go

···
+package media
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/go-gst/go-gst/gst"
+	"github.com/go-gst/go-gst/gst/app"
+	"github.com/google/uuid"
+	"github.com/pion/webrtc/v4"
+	"github.com/pion/webrtc/v4/pkg/media"
+	"stream.place/streamplace/pkg/log"
+	"stream.place/streamplace/pkg/spmetrics"
+)
+
+// we have a bug that prevents us from correctly probing video durations
+// a lot of the time. so when we don't have them we use the last duration
+// that we had, and when we don't have that we use a default duration
+var DEFAULT_DURATION = time.Duration(32 * time.Millisecond)
+
+// This function remains in scope for the duration of a single users' playback
+func (mm *MediaManager) WebRTCPlayback(ctx context.Context, user string, rendition string, offer *webrtc.SessionDescription) (*webrtc.SessionDescription, error) {
+	uu, err := uuid.NewV7()
+	if err != nil {
+		return nil, err
+	}
+	ctx = log.WithLogValues(ctx, "webrtcID", uu.String())
+	ctx, cancel := context.WithCancel(ctx)
+
+	ctx = log.WithLogValues(ctx, "mediafunc", "WebRTCPlayback")
+
+	pipelineSlice := []string{
+		"h264parse name=videoparse ! video/x-h264,stream-format=byte-stream ! appsink name=videoappsink",
+		"opusparse name=audioparse ! appsink name=audioappsink",
+	}
+
+	pipeline, err := gst.NewPipelineFromString(strings.Join(pipelineSlice, "\n"))
+	if err != nil {
+		return nil, fmt.Errorf("failed to create GStreamer pipeline: %w", err)
+	}
+
+	go func() {
+		HandleBusMessages(ctx, pipeline)
+		cancel()
+	}()
+
+	outputQueue, done, err := ConcatStream(ctx, pipeline, user, rendition, mm)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get output queue: %w", err)
+	}
+	go func() {
+		select {
+		case <-ctx.Done():
+			return
+		case <-done:
+			cancel()
+		}
+	}()
+	// queuePadVideo := outputQueue.GetRequestPad("src_%u")
+	// if queuePadVideo == nil {
+	// 	return nil, fmt.Errorf("failed to get queue video pad")
+	// }
+	// queuePadAudio := outputQueue.GetRequestPad("src_%u")
+	// if queuePadAudio == nil {
+	// 	return nil, fmt.Errorf("failed to get queue audio pad")
+	// }
+
+	videoParse, err := pipeline.GetElementByName("videoparse")
+	if err != nil {
+		return nil, fmt.Errorf("failed to get video sink element from pipeline: %w", err)
+	}
+	err = outputQueue.Link(videoParse)
+	if err != nil {
+		return nil, fmt.Errorf("failed to link output queue to video parse: %w", err)
+	}
+
+	audioParse, err := pipeline.GetElementByName("audioparse")
+	if err != nil {
+		return nil, fmt.Errorf("failed to get audio parse element from pipeline: %w", err)
+	}
+	err = outputQueue.Link(audioParse)
+	if err != nil {
+		return nil, fmt.Errorf("failed to link output queue to audio parse: %w", err)
+	}
+
+	videoappsinkele, err := pipeline.GetElementByName("videoappsink")
+	if err != nil {
+		return nil, fmt.Errorf("failed to get video sink element from pipeline: %w", err)
+	}
+
+	audioappsinkele, err := pipeline.GetElementByName("audioappsink")
+	if err != nil {
+		return nil, fmt.Errorf("failed to get audio sink element from pipeline: %w", err)
+	}
+
+	// Create a new RTCPeerConnection
+	peerConnection, err := mm.webrtcAPI.NewPeerConnection(mm.webrtcConfig)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create WebRTC peer connection: %w", err)
+	}
+	go func() {
+		<-ctx.Done()
+		if cErr := peerConnection.Close(); cErr != nil {
+			log.Log(ctx, "cannot close peerConnection: %v\n", cErr)
+		}
+	}()
+
+	videoTrack, err := webrtc.NewTrackLocalStaticSample(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264}, "video", "pion")
+	if err != nil {
+		return nil, fmt.Errorf("failed to create video track: %w", err)
+	}
+	videoRTPSender, err := peerConnection.AddTrack(videoTrack)
+	if err != nil {
+		return nil, fmt.Errorf("failed to add video track to peer connection: %w", err)
+	}
+
+	audioTrack, err := webrtc.NewTrackLocalStaticSample(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus}, "audio", "pion")
+	if err != nil {
+		return nil, fmt.Errorf("failed to create audio track: %w", err)
+	}
+	audioRTPSender, err := peerConnection.AddTrack(audioTrack)
+	if err != nil {
+		return nil, fmt.Errorf("failed to add audio track to peer connection: %w", err)
+	}
+
+	// Set the remote SessionDescription
+	if err = peerConnection.SetRemoteDescription(*offer); err != nil {
+		return nil, fmt.Errorf("failed to set remote description: %w", err)
+	}
+
+	// Create answer
+	answer, err := peerConnection.CreateAnswer(nil)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create answer: %w", err)
+	}
+
+	// Sets the LocalDescription, and starts our UDP listeners
+	if err = peerConnection.SetLocalDescription(answer); err != nil {
+		return nil, fmt.Errorf("failed to set local description: %w", err)
+	}
+
+	// Create channel that is blocked until ICE Gathering is complete
+	gatherComplete := webrtc.GatheringCompletePromise(peerConnection)
+
+	// Setup complete! Now we boot up streaming in the background while returning the SDP offer to the user.
+
+	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)
+			}
+		}
+	}()
+
+	var lastVideoDuration = &DEFAULT_DURATION
+
+	go func() {
+
+		videoappsink := app.SinkFromElement(videoappsinkele)
+		videoappsink.SetCallbacks(&app.SinkCallbacks{
+			NewSampleFunc: func(sink *app.Sink) gst.FlowReturn {
+				sample := sink.PullSample()
+				if sample == nil {
+					return gst.FlowEOS
+				}
+
+				buffer := sample.GetBuffer()
+				if buffer == nil {
+					return gst.FlowError
+				}
+
+				samples := buffer.Map(gst.MapRead).Bytes()
+				defer buffer.Unmap()
+				b2 := make([]byte, len(samples))
+				copy(b2, samples)
+				clockTime := buffer.Duration()
+				dur := clockTime.AsDuration()
+				mediaSample := media.Sample{Data: b2}
+				if dur != nil {
+					mediaSample.Duration = *dur
+					lastVideoDuration = dur
+				} else if lastVideoDuration != nil {
+					mediaSample.Duration = *lastVideoDuration
+				} else {
+					log.Log(ctx, "no video duration", "samples", len(b2))
+					// cancel()
+					return gst.FlowOK
+				}
+
+				if err := videoTrack.WriteSample(mediaSample); err != nil {
+					log.Log(ctx, "failed to write video sample", "error", err)
+					cancel()
+				}
+
+				return gst.FlowOK
+			},
+			EOSFunc: func(sink *app.Sink) {
+				log.Warn(ctx, "videoappsink EOSFunc")
+				cancel()
+			},
+		})
+
+		audioappsink := app.SinkFromElement(audioappsinkele)
+		audioappsink.SetCallbacks(&app.SinkCallbacks{
+			NewSampleFunc: func(sink *app.Sink) gst.FlowReturn {
+				sample := sink.PullSample()
+				if sample == nil {
+					return gst.FlowEOS
+				}
+
+				buffer := sample.GetBuffer()
+				if buffer == nil {
+					return gst.FlowError
+				}
+
+				samples := buffer.Map(gst.MapRead).Bytes()
+				defer buffer.Unmap()
+
+				b2 := make([]byte, len(samples))
+				copy(b2, samples)
+
+				clockTime := buffer.Duration()
+				dur := clockTime.AsDuration()
+				mediaSample := media.Sample{Data: b2}
+				if dur != nil {
+					mediaSample.Duration = *dur
+				} else {
+					log.Log(ctx, "no audio duration", "samples", len(b2))
+					// cancel()
+					return gst.FlowOK
+				}
+				if err := audioTrack.WriteSample(mediaSample); err != nil {
+					log.Log(ctx, "failed to write audio sample", "error", err)
+					return gst.FlowOK
+				}
+
+				return gst.FlowOK
+			},
+			EOSFunc: func(sink *app.Sink) {
+				log.Warn(ctx, "audioappsink EOSFunc")
+				cancel()
+			},
+		})
+
+		// Start the pipeline
+		pipeline.SetState(gst.StatePlaying)
+		spmetrics.ViewerInc(user)
+		defer spmetrics.ViewerDec(user)
+
+		go func() {
+			rtcpBuf := make([]byte, 1500)
+			for {
+				if _, _, rtcpErr := videoRTPSender.Read(rtcpBuf); rtcpErr != nil {
+					return
+				}
+			}
+		}()
+
+		go func() {
+			rtcpBuf := make([]byte, 1500)
+			for {
+				if _, _, rtcpErr := audioRTPSender.Read(rtcpBuf); rtcpErr != nil {
+					return
+				}
+			}
+		}()
+
+		// Set the handler for ICE connection state
+		// This will notify you when the peer has connected/disconnected
+		peerConnection.OnICEConnectionStateChange(func(connectionState webrtc.ICEConnectionState) {
+			log.Log(ctx, "Connection State has changed", "state", connectionState.String())
+			if connectionState == webrtc.ICEConnectionStateConnected {
+				// iceConnectedCtxCancel()
+			}
+		})
+
+		// Set the handler for Peer connection state
+		// This will notify you when the peer has connected/disconnected
+		peerConnection.OnConnectionStateChange(func(s webrtc.PeerConnectionState) {
+			log.Log(ctx, "Peer Connection State has changed", "state", s.String())
+
+			if s == webrtc.PeerConnectionStateFailed || s == webrtc.PeerConnectionStateClosed || s == webrtc.PeerConnectionStateDisconnected {
+				// Wait until PeerConnection has had no network activity for 30 seconds or another failure. It may be reconnected using an ICE Restart.
+				// Use webrtc.PeerConnectionStateDisconnected if you are interested in detecting faster timeout.
+				// Note that the PeerConnection may come back from PeerConnectionStateDisconnected.
+				log.Log(ctx, "Peer Connection has gone to failed, exiting")
+				cancel()
+			}
+		})
+
+		<-ctx.Done()
+
+		log.Warn(ctx, "setting playback pipeline state to null")
+		err = pipeline.BlockSetState(gst.StateNull)
+		if err != nil {
+			log.Log(ctx, "failed to set pipeline state to null", "error", err)
+		}
+
+		videoappsink.SetCallbacks(&app.SinkCallbacks{})
+		err = videoappsinkele.SetState(gst.StateNull)
+		if err != nil {
+			log.Log(ctx, "failed to set videoappsinkele state to null", "error", err)
+		}
+
+		audioappsink.SetCallbacks(&app.SinkCallbacks{})
+		err = audioappsinkele.SetState(gst.StateNull)
+		if err != nil {
+			log.Log(ctx, "failed to set audioappsinkele state to null", "error", err)
+		}
+
+		log.Warn(ctx, "exiting playback")
+
+	}()
+	select {
+	case <-gatherComplete:
+		return peerConnection.LocalDescription(), nil
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}