diff --git a/.vscode/launch.json b/.vscode/launch.json index 4615114a5..32a22c668 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -13,7 +13,7 @@ "type": "go", "request": "launch", "mode": "exec", - "program": "${workspaceFolder}/build-darwin-amd64/libstreamplace" + "program": "${workspaceFolder}/build-darwin-arm64/libstreamplace" } ] } diff --git a/pkg/api/rtmp_server.go b/pkg/api/rtmp_server.go new file mode 100644 index 000000000..858f42ac3 --- /dev/null +++ b/pkg/api/rtmp_server.go @@ -0,0 +1,161 @@ +// Package main contains an example. +package api + +import ( + "context" + "fmt" + "net" + "strings" + "time" + + "github.com/bluenviron/gortmplib" + "github.com/bluenviron/gortsplib/v5/pkg/format" + "golang.org/x/sync/errgroup" + "stream.place/streamplace/pkg/log" + "stream.place/streamplace/pkg/media" +) + +// This example shows how to: +// 1. create a RTMP server +// 2. accept a stream from a reader. +// 3. broadcast the stream to readers. + +// var ( +// mutex sync.Mutex +// publisher *gortmplib.ServerConn +// tracks []format.Format +// readers []*gortmplib.Writer +// ) + +const RTMPPrefix = "/live/" + +func (a *StreamplaceAPI) HandleRTMPPublisher(ctx context.Context, sc *gortmplib.ServerConn) error { + sc.RW.(net.Conn).SetReadDeadline(time.Now().Add(10 * time.Second)) + + if !strings.HasPrefix(sc.URL.Path, RTMPPrefix) { + return fmt.Errorf("RTMP publisher is not allowed to publish to %s (must start with %s)", sc.URL.String(), RTMPPrefix) + } + streamKey := strings.TrimPrefix(sc.URL.Path, RTMPPrefix) + mediaSigner, err := a.MakeMediaSigner(ctx, streamKey) + if err != nil { + return fmt.Errorf("failed to make media signer: %w", err) + } + + ctx = log.WithLogValues(ctx, "streamer", mediaSigner.Streamer()) + + videoInput := make(chan *media.RTMPH264Data, 1024) + defer close(videoInput) + audioInput := make(chan *media.RTMPAACData, 1024) + defer close(audioInput) + + r := &gortmplib.Reader{ + Conn: sc, + } + err = r.Initialize() + if err != nil { + return err + } + + for _, track := range r.Tracks() { + log.Log(ctx, "get track", "track", track) + + switch track := track.(type) { + case *format.H264: + r.OnDataH264(track, func(pts time.Duration, dts time.Duration, au [][]byte) { + log.Log(ctx, "got H264", "len", len(au), "pts", pts, "dts", dts) + videoInput <- &media.RTMPH264Data{ + AU: au, + PTS: pts, + } + }) + + case *format.MPEG4Audio: + r.OnDataMPEG4Audio(track, func(pts time.Duration, au []byte) { + log.Log(ctx, "got MPEG4Au", "len", len(au), "pts", pts) + audioInput <- &media.RTMPAACData{ + AU: au, + PTS: pts, + } + }) + + default: + return fmt.Errorf("unsupported track type: %T", track) + } + } + + g, ctx := errgroup.WithContext(ctx) + g.Go(func() error { + for { + if ctx.Err() != nil { + return ctx.Err() + } + sc.RW.(net.Conn).SetReadDeadline(time.Now().Add(10 * time.Second)) + err = r.Read() + if err != nil { + return err + } + } + }) + + g.Go(func() error { + return a.MediaManager.RTMPIngest(ctx, videoInput, audioInput, mediaSigner) + }) + + return g.Wait() +} + +func (a *StreamplaceAPI) HandleRTMPConnInner(ctx context.Context, conn net.Conn) error { + conn.SetReadDeadline(time.Now().Add(10 * time.Second)) + + sc := &gortmplib.ServerConn{ + RW: conn, + } + err := sc.Initialize() + if err != nil { + return err + } + + err = sc.Accept() + if err != nil { + return err + } + + if sc.Publish { + return a.HandleRTMPPublisher(ctx, sc) + } + return fmt.Errorf("RTMP playback is not supported") +} + +func (a *StreamplaceAPI) HandleRTMPConn(ctx context.Context, conn net.Conn) { + defer conn.Close() + + log.Log(ctx, "connection opened", "remoteAddr", conn.RemoteAddr()) + err := a.HandleRTMPConnInner(ctx, conn) + log.Log(ctx, "connection closed", "remoteAddr", conn.RemoteAddr(), "error", err) +} + +func (a *StreamplaceAPI) StartRTMPServer(ctx context.Context) error { + ln, err := net.Listen("tcp", ":1935") + if err != nil { + return fmt.Errorf("failed to listen: %w", err) + } + defer ln.Close() + + log.Log(ctx, "listening on :1935") + + // Accept loop in a goroutine so we can select on context.Done + go func() { + for { + conn, err := ln.Accept() + if err != nil { + log.Error(ctx, "error accepting RTMP connection", "error", err) + } + + go a.HandleRTMPConn(ctx, conn) + } + }() + + <-ctx.Done() + + return ln.Close() +} diff --git a/pkg/cmd/streamplace.go b/pkg/cmd/streamplace.go index e4935a5ba..1bfea0882 100644 --- a/pkg/cmd/streamplace.go +++ b/pkg/cmd/streamplace.go @@ -448,7 +448,7 @@ func start(build *config.BuildFlags, platformJobs []jobFunc) error { }) group.Go(func() error { - return media.StartRTMPServer(ctx) + return a.StartRTMPServer(ctx) }) group.Go(func() error { diff --git a/pkg/media/rtmp.go b/pkg/media/rtmp.go deleted file mode 100644 index 2b6cb3ca5..000000000 --- a/pkg/media/rtmp.go +++ /dev/null @@ -1,160 +0,0 @@ -// Package main contains an example. -package media - -import ( - "context" - "fmt" - "net" - "time" - - "github.com/bluenviron/gortmplib" - "github.com/bluenviron/gortsplib/v5/pkg/format" - "stream.place/streamplace/pkg/log" -) - -// This example shows how to: -// 1. create a RTMP server -// 2. accept a stream from a reader. -// 3. broadcast the stream to readers. - -// var ( -// mutex sync.Mutex -// publisher *gortmplib.ServerConn -// tracks []format.Format -// readers []*gortmplib.Writer -// ) - -func handlePublisher(ctx context.Context, sc *gortmplib.ServerConn) error { - sc.RW.(net.Conn).SetReadDeadline(time.Now().Add(10 * time.Second)) - - r := &gortmplib.Reader{ - Conn: sc, - } - err := r.Initialize() - if err != nil { - return err - } - - log.Log(ctx, "conn %v is publishing:", sc.RW.(net.Conn).RemoteAddr()) - - for _, track := range r.Tracks() { - log.Log(ctx, "get track", "track", track) - - switch track := track.(type) { - case *format.AV1: - r.OnDataAV1(track, func(pts time.Duration, tu [][]byte) { - log.Log(ctx, "got AV1", "len", len(tu), "pts", pts) - }) - - case *format.VP9: - r.OnDataVP9(track, func(pts time.Duration, frame []byte) { - - log.Log(ctx, "got VP9", "len", len(frame), "pts", pts) - }) - - case *format.H265: - r.OnDataH265(track, func(pts time.Duration, dts time.Duration, au [][]byte) { - log.Log(ctx, "got H265", "len", len(au), "pts", pts, "dts", dts) - }) - - case *format.H264: - r.OnDataH264(track, func(pts time.Duration, dts time.Duration, au [][]byte) { - log.Log(ctx, "got H264", "len", len(au), "pts", pts, "dts", dts) - }) - - case *format.Opus: - r.OnDataOpus(track, func(pts time.Duration, packet []byte) { - log.Log(ctx, "got Opus", "len", len(packet), "pts", pts) - }) - - case *format.MPEG4Audio: - r.OnDataMPEG4Audio(track, func(pts time.Duration, au []byte) { - log.Log(ctx, "got MPEG4Au", "len", len(au), "pts", pts) - }) - - case *format.MPEG1Audio: - r.OnDataMPEG1Audio(track, func(pts time.Duration, frame []byte) { - log.Log(ctx, "got MPEG1Au", "len", len(frame), "pts", pts) - }) - - case *format.AC3: - r.OnDataAC3(track, func(pts time.Duration, frame []byte) { - log.Log(ctx, "got AC3", "len", len(frame), "pts", pts) - }) - - case *format.G711: - r.OnDataG711(track, func(pts time.Duration, samples []byte) { - log.Log(ctx, "got G711", "len", len(samples), "pts", pts) - }) - - case *format.LPCM: - r.OnDataLPCM(track, func(pts time.Duration, samples []byte) { - log.Log(ctx, "got LPCM", "len", len(samples), "pts", pts) - }) - } - } - - for { - sc.RW.(net.Conn).SetReadDeadline(time.Now().Add(10 * time.Second)) - err = r.Read() - if err != nil { - return err - } - } -} - -func handleConnInner(ctx context.Context, conn net.Conn) error { - conn.SetReadDeadline(time.Now().Add(10 * time.Second)) - - sc := &gortmplib.ServerConn{ - RW: conn, - } - err := sc.Initialize() - if err != nil { - return err - } - - err = sc.Accept() - if err != nil { - return err - } - - if sc.Publish { - return handlePublisher(ctx, sc) - } - return fmt.Errorf("RTMP playback is not supported") -} - -func handleConn(ctx context.Context, conn net.Conn) { - defer conn.Close() - - log.Log(ctx, "conn %v opened", conn.RemoteAddr()) - err := handleConnInner(ctx, conn) - log.Log(ctx, "conn %v closed: %v", conn.RemoteAddr(), err) -} - -func StartRTMPServer(ctx context.Context) error { - ln, err := net.Listen("tcp", ":1935") - if err != nil { - return fmt.Errorf("failed to listen: %w", err) - } - defer ln.Close() - - log.Log(ctx, "listening on :1935") - - // Accept loop in a goroutine so we can select on context.Done - go func() { - for { - conn, err := ln.Accept() - if err != nil { - log.Error(ctx, "error accepting RTMP connection", "error", err) - } - - go handleConn(ctx, conn) - } - }() - - <-ctx.Done() - - return ln.Close() -} diff --git a/pkg/media/rtmp_ingest.go b/pkg/media/rtmp_ingest.go new file mode 100644 index 000000000..d101f3b54 --- /dev/null +++ b/pkg/media/rtmp_ingest.go @@ -0,0 +1,161 @@ +package media + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/bluenviron/mediacommon/v2/pkg/codecs/h264" + "github.com/go-gst/go-gst/gst" + "github.com/go-gst/go-gst/gst/app" + "stream.place/streamplace/pkg/log" +) + +type RTMPH264Data struct { + AU [][]byte + PTS time.Duration +} + +type RTMPAACData struct { + AU []byte + PTS time.Duration +} + +// ingest a H264+AAC RTMP stream +func (mm *MediaManager) RTMPIngest(ctx context.Context, videoInput chan *RTMPH264Data, audioInput chan *RTMPAACData, ms MediaSigner) error { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + pipelineSlice := []string{ + "appsrc name=videosrc ! queue ! h264parse name=parse", + "appsrc name=audiosrc ! queue ! fdkaacdec ! audioresample ! opusenc name=audioenc", + } + pipeline, err := gst.NewPipelineFromString(strings.Join(pipelineSlice, "\n")) + if err != nil { + return fmt.Errorf("error creating RTMPIngest pipeline: %w", err) + } + + videosrcEle, err := pipeline.GetElementByName("videosrc") + if err != nil { + return err + } + // defer runtime.KeepAlive(srcele) + videosrc := app.SrcFromElement(videosrcEle) + videosrc.SetCaps(gst.NewCapsFromString("video/x-h264,stream-format=byte-stream")) + videosrc.SetCallbacks(&app.SourceCallbacks{ + NeedDataFunc: func(self *app.Source, length uint) { + if ctx.Err() != nil { + self.EndStream() + return + } + + packet := <-videoInput + if packet == nil { + log.Debug(ctx, "video input closed, ending stream") + self.EndStream() + return + } + + // allBytes := bytes.Buffer{} + // for _, au := range packet.AU { + // allBytes.Write(au) + // } + + avcc, err := h264.AnnexB(packet.AU).Marshal() + if err != nil { + log.Error(ctx, "failed to marshal AVCC", "error", err) + self.Error("failed to marshal AVCC", fmt.Errorf("failed to marshal AVCC: %w", err)) + return + } + + buf := gst.NewBufferFromBytes(avcc) + buf.SetPresentationTimestamp(gst.ClockTime(uint64(packet.PTS.Nanoseconds()))) + ret := self.PushBuffer(buf) + if ret != gst.FlowOK { + log.Error(ctx, "failed to push video buffer", "error", ret.String()) + self.Error("failed to push video buffer", fmt.Errorf("failed to push video buffer: %s", ret.String())) + return + } + }, + }) + + audiosrcEle, err := pipeline.GetElementByName("videosrc") + if err != nil { + return err + } + // defer runtime.KeepAlive(srcele) + audiosrc := app.SrcFromElement(audiosrcEle) + audiosrc.SetCallbacks(&app.SourceCallbacks{ + NeedDataFunc: func(self *app.Source, length uint) { + if ctx.Err() != nil { + self.EndStream() + return + } + packet := <-audioInput + if packet == nil { + log.Debug(ctx, "audio input closed, ending stream") + self.EndStream() + return + } + buf := gst.NewBufferFromBytes(packet.AU) + buf.SetPresentationTimestamp(gst.ClockTime(uint64(packet.PTS.Nanoseconds()))) + ret := self.PushBuffer(buf) + if ret != gst.FlowOK { + log.Error(ctx, "failed to push audio buffer", "error", ret.String()) + self.Error("failed to push audio buffer", fmt.Errorf("failed to push audio buffer: %s", ret.String())) + return + } + }, + }) + + parseEle, err := pipeline.GetElementByName("parse") + if err != nil { + return err + } + + signer, err := mm.SegmentAndSignElem(ctx, ms) + if err != nil { + return err + } + + err = pipeline.Add(signer) + if err != nil { + return err + } + err = parseEle.Link(signer) + if err != nil { + return err + } + audioenc, err := pipeline.GetElementByName("audioenc") + if err != nil { + return err + } + err = audioenc.Link(signer) + if err != nil { + return err + } + + busErr := make(chan error) + go func() { + err := HandleBusMessages(ctx, pipeline) + busErr <- err + }() + + go mm.HandleKeyRevocation(ctx, ms, pipeline) + + err = pipeline.SetState(gst.StatePlaying) + if err != nil { + return err + } + + defer func() { + err := pipeline.SetState(gst.StateNull) + if err != nil { + log.Error(ctx, "error setting pipeline to null state", "error", err) + } + }() + + err = <-busErr + + return err +}