Merge pull request #148 from streamplace/eli/segment-madness

media: improvements to segment memory handling

pkg/media/segchanman/segchanman.go

···
 
 	"go.opentelemetry.io/otel"
 	"stream.place/streamplace/pkg/log"
+	"stream.place/streamplace/pkg/spmetrics"
 )
 
 // it's a segment channel manager, you see
···
 	ch := make(chan *Seg)
 	chs = append(chs, ch)
 	s.segChans[key] = chs
+	spmetrics.SegmentSubscriptionsOpen.WithLabelValues(user, rendition).Set(float64(len(chs)))
 	return ch
 }
 
···
 			break
 		}
 	}
+	spmetrics.SegmentSubscriptionsOpen.WithLabelValues(user, rendition).Set(float64(len(chs)))
 	s.segChans[key] = chs
 }
 

pkg/media/webrtc_ingest.go

···
 
 					gbuf := gst.NewBufferWithSize(int64(len(buf[:i])))
 					gbuf.Map(gst.MapWrite).WriteData(buf[:i])
-					defer gbuf.Unmap()
+					gbuf.Unmap()
 
 					ret := videoSrc.PushBuffer(gbuf)
 					if ret != gst.FlowOK {
···
 
 					gbuf := gst.NewBufferWithSize(int64(len(buf[:i])))
 					gbuf.Map(gst.MapWrite).WriteData(buf[:i])
-					defer gbuf.Unmap()
+					gbuf.Unmap()
 					ret := audioSrc.PushBuffer(gbuf)
 					if ret != gst.FlowOK {
 						log.Log(ctx, "failed to push buffer", "error", ret)

pkg/spmetrics/spmetrics.go

···
 	Help: "number of open websockets",
 })
 
+var SegmentSubscriptionsOpen = promauto.NewGaugeVec(prometheus.GaugeOpts{
+	Name: "streamplace_segment_subscriptions_open",
+	Help: "number of open new segment subscriptions",
+}, []string{"streamer", "rendition"})
+
 func ViewerInc(user string) {
 	go func() {
 		viewersLock.Lock()