Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
4.3 kB · 152 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153package bus
import ( "context" "fmt" "time"
"go.opentelemetry.io/otel" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/spmetrics")
// it's a segment channel manager, you see
type Seg struct { Filepath string Data []byte // presentation flat MP4 Muxl []byte // bare canonical .m4s (blindly concatenatable) PacketizedData *PacketizedSegment Published bool}
// PacketizedSample is one WebRTC-writable sample — a video access unit or an// Opus packet — with its real duration on the source timeline. Carrying the// per-sample duration (rather than dividing the segment evenly) keeps// non-uniform frame spacing intact: an encoder shedding frames under// bandwidth pressure sends bursts and gaps, and respacing those uniformly// rubber-bands the video against the audio.type PacketizedSample struct { Data []byte Duration time.Duration}
type PacketizedSegment struct { Video []PacketizedSample Audio []PacketizedSample Duration time.Duration}
var chanSize = 1024
type SegChan struct { C chan *Seg Context context.Context}
var bufSize = 10
func segChanKey(user string, rendition string) string { return fmt.Sprintf("%s::%s", user, rendition)}
// get a channel to subscribe to new segments for a given user and renditionfunc (b *Bus) SubscribeSegment(ctx context.Context, user string, rendition string) *SegChan { return b.SubscribeSegmentBuf(ctx, user, rendition, 0)}
// get a channel to subscribe to new segments for a given user and rendition,// starting with bufSize cached segments that we already havefunc (b *Bus) SubscribeSegmentBuf(ctx context.Context, user string, rendition string, bufSize int) *SegChan { key := segChanKey(user, rendition) b.segChansMutex.Lock() defer b.segChansMutex.Unlock() chs, ok := b.segChans[key] if !ok { chs = []*SegChan{} b.segChans[key] = chs } // Buffered, and fed in publication order by PublishSegment itself (no // goroutine per segment, which delivered bursts in random order); a // subscriber that falls chanSize segments behind loses segments rather // than holding the publisher. The most recent segments are replayed // into it first so a late subscriber can start at once. ch := make(chan *Seg, chanSize) b.segBufMutex.RLock() defer b.segBufMutex.RUnlock() curBuf, ok := b.segBuf[key] if ok { if bufSize > len(curBuf) { bufSize = len(curBuf) } for i := 0; i < bufSize; i += 1 { ch <- curBuf[len(curBuf)-bufSize+i] } } segChan := &SegChan{C: ch, Context: ctx} chs = append(chs, segChan) b.segChans[key] = chs spmetrics.SegmentSubscriptionsOpen.WithLabelValues(user, rendition).Set(float64(len(chs))) return segChan}
// unsubscribe from a channel for a given user and renditionfunc (b *Bus) UnsubscribeSegment(ctx context.Context, user string, rendition string, ch *SegChan) { key := segChanKey(user, rendition) b.segChansMutex.Lock() defer b.segChansMutex.Unlock() chs, ok := b.segChans[key] if !ok { return } for i, c := range chs { if c == ch { chs = append(chs[:i], chs[i+1:]...) break } } spmetrics.SegmentSubscriptionsOpen.WithLabelValues(user, rendition).Set(float64(len(chs))) b.segChans[key] = chs}
func (b *Bus) PublishSegment(ctx context.Context, user string, rendition string, seg *Seg) { ctx, span := otel.Tracer("signer").Start(ctx, "PublishSegment") defer span.End() key := segChanKey(user, rendition) b.segChansMutex.Lock() defer b.segChansMutex.Unlock() b.segBufMutex.Lock() defer b.segBufMutex.Unlock() curBuf, ok := b.segBuf[key] if !ok { curBuf = []*Seg{} b.segBuf[key] = curBuf } curBuf = append(curBuf, seg) if len(curBuf) > bufSize { curBuf = curBuf[1:] } b.segBuf[key] = curBuf chs, ok := b.segChans[key] if !ok { return } for _, segChan := range chs { select { case segChan.C <- seg: case <-segChan.Context.Done(): default: log.Warn(ctx, "segment subscriber is not keeping up, dropping a segment", "user", user, "rendition", rendition, "behind", len(segChan.C)) } }}
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)}