WiP

livekit · Nov 10, 2023 · ce2b3d3 · ce2b3d3
1 parent fb80ad3
commit ce2b3d3
Show file tree

Hide file tree

Showing 5 changed files with 18 additions and 11 deletions.
diff --git a/go.mod b/go.mod
@@ -1,5 +1,7 @@
 module github.com/livekit/ingress
 
+replace github.com/livekit/server-sdk-go => ../server-sdk-go
+
 go 1.20
 
 replace github.com/tinyzimmer/go-glib => github.com/livekit/go-glib v0.0.0-20230811224737-7bfaa4e57420
@@ -44,7 +46,7 @@ require (
 	github.com/eapache/queue v1.1.0 // indirect
 	github.com/gammazero/deque v0.2.1 // indirect
 	github.com/go-jose/go-jose/v3 v3.0.0 // indirect
-	github.com/go-logr/logr v1.2.4 // indirect
+	github.com/go-logr/logr v1.3.0 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/golang/protobuf v1.5.3 // indirect
 	github.com/google/uuid v1.3.1 // indirect

diff --git a/go.sum b/go.sum
@@ -30,8 +30,8 @@ github.com/gammazero/deque v0.2.1/go.mod h1:LFroj8x4cMYCukHJDbxFCkT+r9AndaJnFMuZ
 github.com/go-jose/go-jose/v3 v3.0.0 h1:s6rrhirfEP/CGIoc6p+PZAeogN2SxKav6Wp7+dyMWVo=
 github.com/go-jose/go-jose/v3 v3.0.0/go.mod h1:RNkWWRld676jZEYoV3+XK8L2ZnNSvIsxFMht0mSX+u8=
 github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
-github.com/go-logr/logr v1.2.4 h1:g01GSCwiDw2xSZfjJ2/T9M+S6pFdcNtFYsp+Y43HYDQ=
-github.com/go-logr/logr v1.2.4/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
+github.com/go-logr/logr v1.3.0 h1:2y3SDp0ZXuc6/cjLSZ+Q3ir+QB9T/iG5yYRXqsagWSY=
+github.com/go-logr/logr v1.3.0/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
 github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
 github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
 github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0/go.mod h1:fyg7847qk6SyHyPtNmDHnmrv/HOrqktSC+C9fM+CJOE=
@@ -92,8 +92,6 @@ github.com/livekit/protocol v1.9.0 h1:YCTvVNxlz36Y3Fsjec7if1+6HpwvZAqAzBfKYBsS5E
 github.com/livekit/protocol v1.9.0/go.mod h1:l2WjlZWErS6vBlQaQyCGwWLt1aOx10XfQTsmvLjJWFQ=
 github.com/livekit/psrpc v0.5.0 h1:g+yYNSs6Y1/vM7UlFkB2s/ARe2y3RKWZhX8ata5j+eo=
 github.com/livekit/psrpc v0.5.0/go.mod h1:1XYH1LLoD/YbvBvt6xg2KQ/J3InLXSJK6PL/+DKmuAU=
-github.com/livekit/server-sdk-go v1.1.2-0.20231109200842-49b1d3c9a294 h1:2SKzTihUAJxSQsv9/dbpdQL/ycNRFqdcoFQXxpygWnI=
-github.com/livekit/server-sdk-go v1.1.2-0.20231109200842-49b1d3c9a294/go.mod h1:jqftpIcT7KAaZphH7VJOCKdIBNu8vEZuQg0r0Tui3EU=
 github.com/mackerelio/go-osstat v0.2.4 h1:qxGbdPkFo65PXOb/F/nhDKpF2nGmGaCFDLXoZjJTtUs=
 github.com/mackerelio/go-osstat v0.2.4/go.mod h1:Zy+qzGdZs3A9cuIqmgbJvwbmLQH9dJvtio5ZjJTbdlQ=
 github.com/magefile/mage v1.15.0 h1:BvGheCMAsG3bWUDbZ8AyXXpCNwU9u5CB6sM+HNb9HYg=

diff --git a/pkg/lksdk_output/lksdk_output.go b/pkg/lksdk_output/lksdk_output.go
@@ -124,10 +124,8 @@ func (s *LKSDKOutput) AddVideoTrack(outputs []VideoSampleProvider, layers []*liv
 	var err error
 	var activeLayerCount int32
 
-	tracks := make([]*lksdk.LocalSampleTrack, 0)
-	for i, layer := range layers {
-		output := outputs[i]
-		onComplete := func() {
+	getOnComplete := func(layer *livekit.VideoLayer, output VideoSampleProvider) func() {
+		return func() {
 			s.logger.Debugw("video track layer write complete", "layer", layer.Quality.String())
 			// don't unpublish if the completion is due to the output closing
 			if pub != nil && !s.closed.IsBroken() {
@@ -140,6 +138,11 @@ func (s *LKSDKOutput) AddVideoTrack(outputs []VideoSampleProvider, layers []*liv
 			}
 			output.Close()
 		}
+	}
+
+	tracks := make([]*lksdk.LocalSampleTrack, 0)
+	for i, layer := range layers {
+		output := outputs[i]
 
 		onRTCP := func(pkt rtcp.Packet) {
 			switch pkt.(type) {
@@ -159,6 +162,7 @@ func (s *LKSDKOutput) AddVideoTrack(outputs []VideoSampleProvider, layers []*liv
 			return err
 		}
 
+		onComplete := getOnComplete(layer, output)
 		track.OnBind(func() {
 			if err := track.StartWrite(output, onComplete); err != nil {
 				s.logger.Errorw("could not start writing video track", err)

diff --git a/pkg/media/output.go b/pkg/media/output.go
@@ -15,6 +15,7 @@
 package media
 
 import (
+	"context"
 	"fmt"
 	"io"
 	"time"
@@ -399,13 +400,14 @@ func (e *Output) writeSample(s *media.Sample, pts time.Duration) error {
 	}
 }
 
-func (e *Output) NextSample() (media.Sample, error) {
+func (e *Output) NextSample(ctx context.Context) (media.Sample, error) {
 	var s *sample
 
 	for {
 		select {
 		case s = <-e.samples:
 		case <-e.fuse.Watch():
+		case <-ctx.Done():
 		}
 
 		if s == nil {

diff --git a/pkg/whip/sdk_media_sink.go b/pkg/whip/sdk_media_sink.go
@@ -97,10 +97,11 @@ func (sp *SDKMediaSink) PushSample(s *media.Sample, ts time.Duration) error {
 	return nil
 }
 
-func (sp *SDKMediaSink) NextSample() (media.Sample, error) {
+func (sp *SDKMediaSink) NextSample(ctx context.Context) (media.Sample, error) {
 	for {
 		select {
 		case <-sp.fuse.Watch():
+		case <-ctx.Done():
 			return media.Sample{}, io.EOF
 		case s := <-sp.readySamples:
 			drop, err := sp.outputSync.WaitForMediaTime(s.ts)