From 180f3a6b1e9f79b98a70c47c94b1ced34b9c5b25 Mon Sep 17 00:00:00 2001 From: Muhammad Hassan Date: Sun, 21 May 2023 18:37:48 +0500 Subject: [PATCH 1/4] update .gitignore --- .gitignore | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/.gitignore b/.gitignore index d0aa4d5..3cb25e2 100644 --- a/.gitignore +++ b/.gitignore @@ -39,4 +39,9 @@ test/*.h264 test/*.ogg test/*.mp4 -private/ \ No newline at end of file +private/ + + +# Files saved by file writer +*.h264 +*.ogg \ No newline at end of file From 102a48400d4d1e6213f12ae25783c438d1a30ffb Mon Sep 17 00:00:00 2001 From: Muhammad Hassan Date: Sun, 21 May 2023 18:40:35 +0500 Subject: [PATCH 2/4] Add ogg file writer scaffolding --- config/config.go | 1 + pkg/control/control.go | 19 +++++++--- pkg/control/disk.go | 18 +-------- pkg/control/ingest.go | 85 ++++++++++++++++++++++++++++++------------ pkg/control/stream.go | 3 ++ pkg/disk/media.go | 34 ++++++++++++++++- 6 files changed, 113 insertions(+), 47 deletions(-) diff --git a/config/config.go b/config/config.go index 8b7f8ec..edd7923 100644 --- a/config/config.go +++ b/config/config.go @@ -75,6 +75,7 @@ type Config struct { HTTPSKey string `fig:"https_key"` SaveVideo bool `fig:"save_video"` + SaveAudio bool `fig:"save_audio"` } } diff --git a/pkg/control/control.go b/pkg/control/control.go index 70eb4f9..1e64329 100644 --- a/pkg/control/control.go +++ b/pkg/control/control.go @@ -45,9 +45,10 @@ type Control struct { HTTPSCert string `mapstructure:"https_cert"` HTTPSKey string `mapstructure:"https_key"` - // Flag to enable saving video stream to file + // Flag to enable saving audio/video stream to file // Currently it's global flag toggled from the config file SaveVideo bool `mapstructure:"save_video"` + SaveAudio bool `mapstructure:"save_audio"` } func New( @@ -87,8 +88,9 @@ func New( HTTPSCert: httpCfg.HTTPSCert, HTTPSKey: httpCfg.HTTPSKey, - // this should be controlled at a stream level + // these should be controlled at a stream level SaveVideo: cfg.Control.SaveVideo, + SaveAudio: cfg.Control.SaveAudio, }, nil } @@ -358,7 +360,7 @@ func (ctrl *Control) sendThumbnail(channelID types.ChannelID) (err error) { } func (ctrl *Control) newStream(channelID types.ChannelID, cancelFunc context.CancelFunc) (*Stream, error) { - stream := &Stream{ //nolint exhaustive struct + stream := &Stream{ //nolint ChannelID: channelID, log: ctrl.log.WithField("channel_id", channelID), @@ -372,18 +374,25 @@ func (ctrl *Control) newStream(channelID types.ChannelID, cancelFunc context.Can stopThumbnailer: make(chan struct{}, 1), thumbnailReceiver: make(chan *rtp.Packet, 50), requestThumbnail: make(chan struct{}, 1), + videoWriter: &noopFileWriter{}, + audioWriter: &noopFileWriter{}, lastThumbnail: make(chan []byte, 1), startTime: time.Now().Unix(), } - // TODO: this shouldn't be a global flag - // but rather configured on per stream basis + // TODO1: this shouldn't be a global flag + // TODO2: but rather configured on per stream basis + // TODO3: this looks messy and there's probably a better way of doing it if ctrl.SaveVideo { stream.saveVideo = true stream.videoWriterChan = make(chan *rtp.Packet, 100) // not sure what the buffer size here should be } + if ctrl.SaveAudio { + stream.saveAudio = true + stream.audioWriterChan = make(chan *rtp.Packet, 100) // not sure what the buffer size here should be + } if _, exists := ctrl.streams[channelID]; exists { return stream, errors.New("stream already exists in stream manager state") diff --git a/pkg/control/disk.go b/pkg/control/disk.go index 01946e5..519e4f4 100644 --- a/pkg/control/disk.go +++ b/pkg/control/disk.go @@ -2,6 +2,7 @@ package control import ( "github.com/Glimesh/waveguide/pkg/disk" + "github.com/pion/rtp" "github.com/sirupsen/logrus" ) @@ -23,23 +24,6 @@ type fileWriter struct { writer disk.Writer packetCh chan *rtp.Packet done chan struct{} - codec string -} - -func NewVideoWriter(log logrus.FieldLogger, codec, filename string) *fileWriter { - writer, err := disk.NewVideoWriter(codec, filename) - // TODO: handle error - if err != nil { - panic(err) - } - - return &fileWriter{ - writer: writer, - codec: codec, - log: log, - packetCh: make(chan *rtp.Packet, 100), - done: make(chan struct{}, 1), - } } func (fw *fileWriter) SendRTP(p *rtp.Packet) { diff --git a/pkg/control/ingest.go b/pkg/control/ingest.go index 32af23d..30a747c 100644 --- a/pkg/control/ingest.go +++ b/pkg/control/ingest.go @@ -8,6 +8,8 @@ import ( "net/http" "strings" + "github.com/Glimesh/waveguide/pkg/disk" + "github.com/pion/rtp" "github.com/pion/webrtc/v3" ) @@ -20,42 +22,46 @@ func (s *Stream) Ingest(ctx context.Context) error { return err } - s.videoWriter = &noopFileWriter{} - go s.startVideoIngestor() pc.OnTrack(func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) { cancelRead := make(chan struct{}, 1) go func() { + // this goroutine spawns each time for audio & video track + // for now it leaking as there's no way to cancel it + // for the audio track + // TODO: fix goroutine leak for audio track + s.log.Debug("starting cancel read loop") <-done s.videoWriter.Done() - s.log.Debug("exiting on track") LOOP: for { select { + // drain the thumbnail channel on exit case <-s.lastThumbnail: default: s.log.Debug("thumbnail channel drained") break LOOP } } + // cancel the read loop and stop the ingestion + s.log.Debug("exiting cancel read loop") cancelRead <- struct{}{} }() - codec := track.Codec() - - if trackCodec := codec.MimeType; trackCodec == webrtc.MimeTypeH264 { - if s.saveVideo { - filename := fmt.Sprintf("stream.%d.%s", s.StreamID, "out.h264") - videoFileWriter := NewVideoWriter( - s.log.WithField("file-writer", webrtc.MimeTypeH264), - webrtc.MimeTypeH264, - filename, - ) - s.videoWriter = videoFileWriter + codec := track.Codec().MimeType + kind := track.Kind() - go videoFileWriter.Run() + // only h264 codec is supported for now, hence the check below + // this will probably have to go away once support for other codecs is added + if trackCodec := codec; trackCodec == webrtc.MimeTypeH264 { + s.log.Debug("got h264 track") + writer, err := s.initFileWriter(codec, kind) //nolint no shadow + if err != nil { + s.log.Debugf("failed to init video file writer: %v", err) } + s.videoWriter = writer + go s.videoWriter.Run() for { select { @@ -64,18 +70,21 @@ func (s *Stream) Ingest(ctx context.Context) error { close(s.rtpIngest) return default: - pkt, _, readErr := track.ReadRTP() - if readErr != nil { - // terminate the ingestor is input stream is EOF - if errors.Is(readErr, io.EOF) { - s.log.Debugf("read: %v", readErr) - close(s.rtpIngest) - return - } + } + pkt, _, readErr := track.ReadRTP() + if readErr != nil { + // terminate the ingestor is input stream is EOF + if errors.Is(readErr, io.EOF) { + s.log.Debugf("read: %v", readErr) + close(s.rtpIngest) + return } - s.rtpIngest <- pkt } + s.rtpIngest <- pkt } + } else if trackCodec == webrtc.MimeTypeOpus { + // TODO: implement opus writer + s.log.Debug("got opus track") } }) @@ -134,6 +143,34 @@ func (s *Stream) Ingest(ctx context.Context) error { return nil } +func (s *Stream) initFileWriter(mime string, kind webrtc.RTPCodecType) (FileWriter, error) { + var ( + writer disk.Writer + err error + streamID = fmt.Sprintf("%d", s.StreamID) + ) + switch kind { + case webrtc.RTPCodecTypeVideo: + if s.saveVideo { + writer, err = disk.NewVideoWriter(mime, streamID) + } + case webrtc.RTPCodecTypeAudio: + if s.saveAudio { + writer, err = disk.NewAudioWriter(mime, streamID) + } + } + if err != nil { + return nil, err + } + + return &fileWriter{ + writer: writer, + log: s.log.WithField("file-writer", mime), + packetCh: make(chan *rtp.Packet, 100), + done: make(chan struct{}, 1), + }, nil +} + func (s *Stream) startVideoIngestor() { doneThumb := make(chan struct{}, 1) diff --git a/pkg/control/stream.go b/pkg/control/stream.go index 66b0294..4559c5f 100644 --- a/pkg/control/stream.go +++ b/pkg/control/stream.go @@ -30,7 +30,9 @@ type Stream struct { whepURI string saveVideo bool + saveAudio bool videoWriter FileWriter + audioWriter FileWriter // mediaStarted is set after media bytes have come in from the client mediaStarted bool @@ -47,6 +49,7 @@ type Stream struct { thumbnailReceiver chan *rtp.Packet videoWriterChan chan *rtp.Packet + audioWriterChan chan *rtp.Packet ChannelID types.ChannelID StreamID types.StreamID diff --git a/pkg/disk/media.go b/pkg/disk/media.go index dcc5903..8ccde3b 100644 --- a/pkg/disk/media.go +++ b/pkg/disk/media.go @@ -2,11 +2,13 @@ package disk import ( "errors" + "fmt" "github.com/pion/rtp" "github.com/pion/webrtc/v3" "github.com/pion/webrtc/v3/pkg/media" "github.com/pion/webrtc/v3/pkg/media/h264writer" + "github.com/pion/webrtc/v3/pkg/media/oggwriter" ) // TODO: figure out if wrapping this is needed @@ -24,7 +26,12 @@ func (w *writer) Close() error { return w.Writer.Close() } -func NewVideoWriter(codec, filename string) (Writer, error) { +// NewVideoWriter returns a new video writer for the supplied codec +// The writer write the raw encoded stream to disk with the stream ID +// prepended to the codec name e.g. for an h264 stream id "foo" +// the filename is "foo.h264" +func NewVideoWriter(codec, streamID string) (Writer, error) { + filename := fmt.Sprintf("%s.h264", streamID) switch codec { case webrtc.MimeTypeH264: w, err := h264writer.New(filename) @@ -39,3 +46,28 @@ func NewVideoWriter(codec, filename string) (Writer, error) { return nil, errors.New("unsupported codec") } } + +const ( + audioSampleRate = 48000 + audioChannelCount = 2 +) + +// NewAudioWriter returns a new audio writer for the supplied codec +// The writer write the raw encoded stream to disk with the stream ID +// prepended to the codec name e.g. for an ogg stream id "foo" +// the filename is "foo.ogg" +func NewAudioWriter(codec, streamID string) (Writer, error) { + filename := fmt.Sprintf("%s.ogg", streamID) + switch codec { + case webrtc.MimeTypeOpus: + w, err := oggwriter.New(filename, audioSampleRate, audioChannelCount) + if err != nil { + return nil, err + } + return &writer{w}, nil + // TODO: support more audio codec? + // case + default: + return nil, errors.New("unsupported codec") + } +} From 4fd7f842a1319395ff4bb86953691baab4601336 Mon Sep 17 00:00:00 2001 From: Muhammad Hassan Date: Sun, 21 May 2023 19:35:16 +0500 Subject: [PATCH 3/4] Update config example --- config.toml.example | 1 + 1 file changed, 1 insertion(+) diff --git a/config.toml.example b/config.toml.example index f6effad..62db6f9 100644 --- a/config.toml.example +++ b/config.toml.example @@ -28,3 +28,4 @@ orchestrator = "dummy" http_server_type = "http" http_address = "localhost:8091" save_video = false +save_audio = false From f4163fee246e89e5111ca15add121bfc5eca8830 Mon Sep 17 00:00:00 2001 From: Muhammad Hassan Date: Sun, 21 May 2023 21:01:23 +0500 Subject: [PATCH 4/4] Implement ogg writer --- go.mod | 19 ++-- go.sum | 34 +++++--- pkg/control/control.go | 3 +- pkg/control/ingest.go | 193 ++++++++++++++++++++++++++++------------- pkg/control/stream.go | 7 +- 5 files changed, 167 insertions(+), 89 deletions(-) diff --git a/go.mod b/go.mod index a7e9190..92af87b 100644 --- a/go.mod +++ b/go.mod @@ -9,24 +9,25 @@ require ( github.com/google/uuid v1.3.0 github.com/hasura/go-graphql-client v0.8.1 github.com/kkyr/fig v0.3.1 + github.com/livekit/server-sdk-go v1.0.10 github.com/nareix/joy5 v0.0.0-20210317075623-2c912ca30590 github.com/pion/interceptor v0.1.12 github.com/pion/rtcp v1.2.10 github.com/pion/rtp v1.7.13 - github.com/pion/webrtc/v3 v3.1.56 + github.com/pion/webrtc/v3 v3.1.59 github.com/pkg/errors v0.9.1 github.com/sirupsen/logrus v1.9.0 - github.com/stretchr/testify v1.8.1 + github.com/stretchr/testify v1.8.2 github.com/yutopp/go-flv v0.2.0 github.com/yutopp/go-rtmp v0.0.1 - golang.org/x/crypto v0.6.0 + golang.org/x/crypto v0.8.0 golang.org/x/oauth2 v0.1.0 gopkg.in/hraban/opus.v2 v2.0.0-20220302220929-eeacdbcb92d0 ) require ( github.com/davecgh/go-spew v1.1.1 // indirect - github.com/golang/protobuf v1.5.2 // indirect + github.com/golang/protobuf v1.5.3 // indirect github.com/google/go-cmp v0.5.9 // indirect github.com/hashicorp/errwrap v1.1.0 // indirect github.com/hashicorp/go-multierror v1.1.0 // indirect @@ -39,7 +40,7 @@ require ( github.com/pelletier/go-toml v1.9.5 // indirect github.com/pion/datachannel v1.5.5 // indirect github.com/pion/dtls/v2 v2.2.6 // indirect - github.com/pion/ice/v2 v2.3.1 // indirect + github.com/pion/ice/v2 v2.3.2 // indirect github.com/pion/logging v0.2.2 // indirect github.com/pion/mdns v0.0.7 // indirect github.com/pion/randutil v0.1.0 // indirect @@ -52,11 +53,11 @@ require ( github.com/pion/udp/v2 v2.0.1 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect github.com/yutopp/go-amf0 v0.0.0-20180803120851-48851794bb1f // indirect - golang.org/x/net v0.7.0 // indirect - golang.org/x/sys v0.5.0 // indirect - golang.org/x/text v0.7.0 // indirect + golang.org/x/net v0.9.0 // indirect + golang.org/x/sys v0.7.0 // indirect + golang.org/x/text v0.9.0 // indirect google.golang.org/appengine v1.6.7 // indirect - google.golang.org/protobuf v1.28.1 // indirect + google.golang.org/protobuf v1.30.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect nhooyr.io/websocket v1.8.7 // indirect ) diff --git a/go.sum b/go.sum index bb896e0..7a82754 100644 --- a/go.sum +++ b/go.sum @@ -42,8 +42,9 @@ github.com/golang/protobuf v1.4.0-rc.4.0.20200313231945-b860323f09d0/go.mod h1:W github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvqG2KuDX0= github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI= github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk= -github.com/golang/protobuf v1.5.2 h1:ROPKBNFfQgOUMifHyP+KYbvpjbdoFNs+aK7DXlji0Tw= github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY= +github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg= +github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY= github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= @@ -54,8 +55,8 @@ github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeN github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I= github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/gorilla/websocket v1.4.1 h1:q7AeDBpnBk8AogcD4DSag/Ukw/KV+YhzLj2bP5HvKCM= github.com/gorilla/websocket v1.4.1/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= +github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc= github.com/graph-gophers/graphql-go v1.4.0 h1:JE9wveRTSXwJyjdRd6bOQ7Ob5bewTUQ58Jv4OiVdpdE= github.com/graph-gophers/graphql-go v1.4.0/go.mod h1:YtmJZDLbF1YYNrlNAuiO5zAStUWc3XZT07iGsVqe1Os= github.com/graph-gophers/graphql-transport-ws v0.0.2 h1:DbmSkbIGzj8SvHei6n8Mh9eLQin8PtA8xY9eCzjRpvo= @@ -85,6 +86,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/leodido/go-urn v1.2.0 h1:hpXL4XnriNwQ/ABnpepYM/1vCLWNDfUNts8dX3xTG6Y= github.com/leodido/go-urn v1.2.0/go.mod h1:+8+nEpDfqqsY+g338gtMEUOtuK+4dEMhiQEgxpxOKII= +github.com/livekit/server-sdk-go v1.0.10 h1:v0tL30doD6Odt43Gy4560r2DE1FEA/5IpQZP3yov0Rg= +github.com/livekit/server-sdk-go v1.0.10/go.mod h1:2Zu9YCekLoPbGWrbWyc6+EIJiQwmEs5WbANOrU7pl/8= github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU= github.com/mattn/go-isatty v0.0.14 h1:yVuAays6BHfxijgZPzw+3Zlu5yQgKGP2/hcQbHb7S9Y= github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27kJ6hsGG94= @@ -115,8 +118,8 @@ github.com/pion/datachannel v1.5.5 h1:10ef4kwdjije+M9d7Xm9im2Y3O6A6ccQb0zcqZcJew github.com/pion/datachannel v1.5.5/go.mod h1:iMz+lECmfdCMqFRhXhcA/219B0SQlbpoR2V118yimL0= github.com/pion/dtls/v2 v2.2.6 h1:yXMxKr0Skd+Ub6A8UqXTRLSywskx93ooMRHsQUtd+Z4= github.com/pion/dtls/v2 v2.2.6/go.mod h1:t8fWJCIquY5rlQZwA2yWxUS1+OCrAdXrhVKXB5oD/wY= -github.com/pion/ice/v2 v2.3.1 h1:FQCmUfZe2Jpe7LYStVBOP6z1DiSzbIateih3TztgTjc= -github.com/pion/ice/v2 v2.3.1/go.mod h1:aq2kc6MtYNcn4XmMhobAv6hTNJiHzvD0yXRz80+bnP8= +github.com/pion/ice/v2 v2.3.2 h1:vh+fi4RkZ8H5fB4brZ/jm3j4BqFgMmNs+aB3X52Hu7M= +github.com/pion/ice/v2 v2.3.2/go.mod h1:AMIpuJqcpe+UwloocNebmTSWhCZM1TUCo9v7nW50jX0= github.com/pion/interceptor v0.1.12 h1:CslaNriCFUItiXS5o+hh5lpL0t0ytQkFnUcbbCs2Zq8= github.com/pion/interceptor v0.1.12/go.mod h1:bDtgAD9dRkBZpWHGKaoKb42FhDHTG2rX8Ii9LRALLVA= github.com/pion/logging v0.2.2 h1:M9+AIj/+pxNsDfAT64+MAVgJO0rsyLnoJKCqf//DoeY= @@ -147,8 +150,8 @@ github.com/pion/turn/v2 v2.1.0 h1:5wGHSgGhJhP/RpabkUb/T9PdsAjkGLS6toYz5HNzoSI= github.com/pion/turn/v2 v2.1.0/go.mod h1:yrT5XbXSGX1VFSF31A3c1kCNB5bBZgk/uu5LET162qs= github.com/pion/udp/v2 v2.0.1 h1:xP0z6WNux1zWEjhC7onRA3EwwSliXqu1ElUZAQhUP54= github.com/pion/udp/v2 v2.0.1/go.mod h1:B7uvTMP00lzWdyMr/1PVZXtV3wpPIxBRd4Wl6AksXn8= -github.com/pion/webrtc/v3 v3.1.56 h1:ScaiqKQN3liQwT+kJwOBaYP6TwSfixzdUnZmzHAo0a0= -github.com/pion/webrtc/v3 v3.1.56/go.mod h1:7VhbA6ihqJlz6R/INHjyh1b8HpiV9Ct4UQvE1OB/xoM= +github.com/pion/webrtc/v3 v3.1.59 h1:B3YFo8q6dwBYKA2LUjWRChP59Qtt+xvv1Ul7UPDp6Zc= +github.com/pion/webrtc/v3 v3.1.59/go.mod h1:rJGgStRoFyFOWJULHLayaimsG+jIEoenhJ5MB5gIFqw= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= @@ -171,8 +174,9 @@ github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= -github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk= github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= +github.com/stretchr/testify v1.8.2 h1:+h33VjcLVPDHtOdpUCuF+7gSuG3yGIftsP1YvFihtJ8= +github.com/stretchr/testify v1.8.2/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= github.com/ugorji/go v1.1.7 h1:/68gy2h+1mWMrwZFeD1kQialdSzAb432dtpeJ42ovdo= github.com/ugorji/go v1.1.7/go.mod h1:kZn38zHttfInRq0xu/PH0az30d+z6vm202qpg1oXVMw= github.com/ugorji/go/codec v1.1.7 h1:2SvQaVZ1ouYrrKKwoSk2pzd4A9evlKJb9oTL+OaLUSs= @@ -190,8 +194,9 @@ golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8U golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= golang.org/x/crypto v0.5.0/go.mod h1:NK/OQwhpMQP3MwtdjgLlYHnH9ebylxKWv3e0fK+mkQU= -golang.org/x/crypto v0.6.0 h1:qfktjS5LUO+fFKeJXZ+ikTRijMmljikvG68fpMMruSc= golang.org/x/crypto v0.6.0/go.mod h1:OFC/31mSvZgRz0V1QTNCzfAI1aIRzbiufJtkMIlEp58= +golang.org/x/crypto v0.8.0 h1:pd9TJtTueMTVQXzk8E2XESSMQDj/U7OUu0PqJqPXQjQ= +golang.org/x/crypto v0.8.0/go.mod h1:mRqEX+O9/h5TFCrQhkgjo2yKi0yYA+9ecGkdQoHrywE= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= @@ -207,8 +212,9 @@ golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug golang.org/x/net v0.1.0/go.mod h1:Cx3nUiGt4eDBEyega/BKRp+/AlGL8hYe7U9odMt2Cco= golang.org/x/net v0.5.0/go.mod h1:DivGGAXEgPSlEBzxGzZI+ZLohi+xUj054jfeKui00ws= golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= -golang.org/x/net v0.7.0 h1:rJrUqqhjsgNp7KqAIc25s9pZnjU7TUcSY7HcVZjdn1g= golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= +golang.org/x/net v0.9.0 h1:aWJ/m6xSmxWBx+V0XRHTlrYrPG56jKsLdTFmsSsCzOM= +golang.org/x/net v0.9.0/go.mod h1:d48xBJpPfHeWQsugry2m+kC02ZBRGRgulfHnEXEuWns= golang.org/x/oauth2 v0.1.0 h1:isLCZuhj4v+tYv7eskaN4v/TM+A1begWWgyVJDdl1+Y= golang.org/x/oauth2 v0.1.0/go.mod h1:G9FE4dLTsbXUu90h/Pf85g4w1D+SSAgR+q46nJZ8M4A= golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= @@ -237,8 +243,9 @@ golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.2.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.4.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.5.0 h1:MUK/U/4lj1t1oPg0HfuXDN/Z1wv31ZJ/YcPiGccS4DU= golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.7.0 h1:3jlCCIQZPdOYu1h8BkNvLz8Kgwtae2cagcG/VamtZRU= +golang.org/x/sys v0.7.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= golang.org/x/term v0.1.0/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= @@ -251,8 +258,9 @@ golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= golang.org/x/text v0.6.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= -golang.org/x/text v0.7.0 h1:4BRB4x83lYWy72KwLD/qYDuTu7q9PjSagHvijDw7cLo= golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= +golang.org/x/text v0.9.0 h1:2sjJmO8cDvYveuX97RDLsxlyUxLl+GHoLxBiRdHllBE= +golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8= golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= @@ -272,8 +280,8 @@ google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzi google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU= google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= -google.golang.org/protobuf v1.28.1 h1:d0NfwRgPtno5B1Wa6L2DAG+KivqkdutMf1UhdNx175w= -google.golang.org/protobuf v1.28.1/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I= +google.golang.org/protobuf v1.30.0 h1:kPPoIgf3TsEvrm0PFe15JQ+570QVxYzEvvHqChK+cng= +google.golang.org/protobuf v1.30.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo= diff --git a/pkg/control/control.go b/pkg/control/control.go index 1e64329..1aeb03a 100644 --- a/pkg/control/control.go +++ b/pkg/control/control.go @@ -369,13 +369,12 @@ func (ctrl *Control) newStream(channelID types.ChannelID, cancelFunc context.Can cancelFunc: cancelFunc, kf: keyframer.New(), - rtpIngest: make(chan *rtp.Packet), + videoRTPIngest: make(chan *rtp.Packet), stopHeartbeat: make(chan struct{}, 1), stopThumbnailer: make(chan struct{}, 1), thumbnailReceiver: make(chan *rtp.Packet, 50), requestThumbnail: make(chan struct{}, 1), videoWriter: &noopFileWriter{}, - audioWriter: &noopFileWriter{}, lastThumbnail: make(chan []byte, 1), diff --git a/pkg/control/ingest.go b/pkg/control/ingest.go index 30a747c..eaf50fc 100644 --- a/pkg/control/ingest.go +++ b/pkg/control/ingest.go @@ -9,13 +9,19 @@ import ( "strings" "github.com/Glimesh/waveguide/pkg/disk" + + "github.com/livekit/server-sdk-go/pkg/samplebuilder" "github.com/pion/rtp" + "github.com/pion/rtp/codecs" "github.com/pion/webrtc/v3" + "github.com/pion/webrtc/v3/pkg/media" + "github.com/pion/webrtc/v3/pkg/media/oggwriter" ) func (s *Stream) Ingest(ctx context.Context) error { logger := s.log.WithField("app", "ingest") - done := make(chan struct{}, 1) + doneVideo := make(chan struct{}, 1) + doneAudio := make(chan struct{}, 1) pc, err := webrtc.NewPeerConnection(webrtc.Configuration{}) //nolint exhaustive struct if err != nil { @@ -24,67 +30,67 @@ func (s *Stream) Ingest(ctx context.Context) error { go s.startVideoIngestor() + // TODO: refactor the following OnTrack callback logic pc.OnTrack(func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) { - cancelRead := make(chan struct{}, 1) - go func() { - // this goroutine spawns each time for audio & video track - // for now it leaking as there's no way to cancel it - // for the audio track - // TODO: fix goroutine leak for audio track - s.log.Debug("starting cancel read loop") - <-done - s.videoWriter.Done() - LOOP: - for { - select { - // drain the thumbnail channel on exit - case <-s.lastThumbnail: - default: - s.log.Debug("thumbnail channel drained") - break LOOP - } - } - // cancel the read loop and stop the ingestion - s.log.Debug("exiting cancel read loop") - cancelRead <- struct{}{} - }() - codec := track.Codec().MimeType kind := track.Kind() // only h264 codec is supported for now, hence the check below // this will probably have to go away once support for other codecs is added + + // refactoring idea for future me + // one approach is to construct ingestor structs for audio and video media + // the respective ingestor holds the reference to the remote track + // and the appropriate sub-consumers e.g. the video ingestor holds the thumbnailer + // and the file-writer while the audio ingestor holds just the file writer + // + // the respective ingestor then holds the logic for multiplexing/writing + // the rtp packets from the remote track to its consumers if trackCodec := codec; trackCodec == webrtc.MimeTypeH264 { s.log.Debug("got h264 track") - writer, err := s.initFileWriter(codec, kind) //nolint no shadow - if err != nil { - s.log.Debugf("failed to init video file writer: %v", err) + cancelRead := s.cancelVideoRead(doneVideo) + if s.saveVideo { + err := s.initFileWriter(codec, kind) //nolint no shadow + if err != nil { + s.log.Debugf("failed to init video file writer: %v", err) + } + go s.videoWriter.Run() } - s.videoWriter = writer - go s.videoWriter.Run() for { select { case <-cancelRead: - s.log.Debug("on track stop signal") - close(s.rtpIngest) + s.log.Debug("on video track stop signal") + close(s.videoRTPIngest) return default: } pkt, _, readErr := track.ReadRTP() - if readErr != nil { + if readErr != nil && errors.Is(readErr, io.EOF) { // terminate the ingestor is input stream is EOF - if errors.Is(readErr, io.EOF) { - s.log.Debugf("read: %v", readErr) - close(s.rtpIngest) - return - } + s.log.Debugf("read: %v", readErr) + close(s.videoRTPIngest) + return } - s.rtpIngest <- pkt + s.videoRTPIngest <- pkt } } else if trackCodec == webrtc.MimeTypeOpus { - // TODO: implement opus writer s.log.Debug("got opus track") + if s.saveAudio { + cancelRead := s.cancelAudioRead(doneAudio) + s.log.Debug("audio file writer enabled") + sb := samplebuilder.New(10, &codecs.OpusPacket{}, track.Codec().ClockRate) + writer, err := oggwriter.New(s.StreamID.String()+".ogg", 48000, track.Codec().Channels) + if err != nil { + return + } + t := &TrackWriter{ + writer: writer, + sb: sb, + track: track, + } + go t.start(cancelRead) + } } }) @@ -136,47 +142,113 @@ func (s *Stream) Ingest(ctx context.Context) error { } <-ctx.Done() - pc.Close() - done <- struct{}{} logger.Debug("received ctx done signal") + pc.Close() + doneVideo <- struct{}{} + doneAudio <- struct{}{} return nil } -func (s *Stream) initFileWriter(mime string, kind webrtc.RTPCodecType) (FileWriter, error) { +type TrackWriter struct { + sb *samplebuilder.SampleBuilder + writer media.Writer + track *webrtc.TrackRemote +} + +func (t *TrackWriter) start(cancel chan struct{}) { + defer t.writer.Close() + for { + select { + case <-cancel: + return + default: + } + pkt, _, err := t.track.ReadRTP() + if err != nil { + break + } + t.sb.Push(pkt) + + for _, p := range t.sb.PopPackets() { + t.writer.WriteRTP(p) + } + } +} + +func (s *Stream) cancelAudioRead(done chan struct{}) chan struct{} { + cancel := make(chan struct{}, 1) + go func() { + s.log.Debug("starting cancel audio read loop") + <-done + // cancel the read loop and stop the ingestion + cancel <- struct{}{} + s.log.Debug("exiting cancel audio read loop") + }() + return cancel +} + +func (s *Stream) cancelVideoRead(done chan struct{}) chan struct{} { + cancel := make(chan struct{}, 1) + go func() { + s.log.Debug("starting cancel video read loop") + <-done + s.videoWriter.Done() + LOOP: + for { + select { + // drain the thumbnail channel on exit + case <-s.lastThumbnail: + default: + s.log.Debug("thumbnail channel drained") + break LOOP + } + } + // cancel the read loop and stop the ingestion + s.log.Debug("exiting cancel video read loop") + cancel <- struct{}{} + }() + return cancel +} + +// initFileWriter initializes the file writer and sets it for the +// stream based on the codec mime and type +func (s *Stream) initFileWriter(mime string, kind webrtc.RTPCodecType) error { var ( writer disk.Writer err error streamID = fmt.Sprintf("%d", s.StreamID) + fw = &fileWriter{ //nolint + log: s.log.WithField("file-writer", mime), + packetCh: make(chan *rtp.Packet, 100), + done: make(chan struct{}, 1), + } ) + switch kind { case webrtc.RTPCodecTypeVideo: - if s.saveVideo { - writer, err = disk.NewVideoWriter(mime, streamID) + writer, err = disk.NewVideoWriter(mime, streamID) + if err != nil { + return err } + fw.writer = writer + s.videoWriter = fw + return nil + + // not needed? case webrtc.RTPCodecTypeAudio: - if s.saveAudio { - writer, err = disk.NewAudioWriter(mime, streamID) - } - } - if err != nil { - return nil, err - } - return &fileWriter{ - writer: writer, - log: s.log.WithField("file-writer", mime), - packetCh: make(chan *rtp.Packet, 100), - done: make(chan struct{}, 1), - }, nil + default: + s.log.Panicf("unknown codec type: %v", kind) + } + return nil } func (s *Stream) startVideoIngestor() { doneThumb := make(chan struct{}, 1) - go s.thumbnailer(doneThumb) - for p := range s.rtpIngest { + for p := range s.videoRTPIngest { select { case s.thumbnailReceiver <- p.Clone(): default: @@ -184,7 +256,6 @@ func (s *Stream) startVideoIngestor() { s.videoWriter.SendRTP(p) } - s.log.Debug("closed ingestor listener") doneThumb <- struct{}{} s.log.Debug("ending rtp ingestor") diff --git a/pkg/control/stream.go b/pkg/control/stream.go index 4559c5f..ed59a66 100644 --- a/pkg/control/stream.go +++ b/pkg/control/stream.go @@ -32,16 +32,15 @@ type Stream struct { saveVideo bool saveAudio bool videoWriter FileWriter - audioWriter FileWriter // mediaStarted is set after media bytes have come in from the client mediaStarted bool hasSomeAudio bool hasSomeVideo bool - kf *keyframer.Keyframer - rtpIngest chan *rtp.Packet - lastThumbnail chan []byte + kf *keyframer.Keyframer + videoRTPIngest chan *rtp.Packet + lastThumbnail chan []byte // channel used to signal thumbnailer to stop stopThumbnailer chan struct{} stopHeartbeat chan struct{}