From d92bb1d4658dccd2b5ba81fe51eb85f529b5b853 Mon Sep 17 00:00:00 2001 From: Yohann D'ANELLO Date: Sun, 8 Nov 2020 00:08:52 +0100 Subject: [PATCH] Try to use manually FFMPEG bindings to avoid having too much syscalls, not working yet --- go.mod | 1 + go.sum | 2 + stream/webrtc/ingest.go | 141 +++++++++++++++++++++++++++++++++------- 3 files changed, 119 insertions(+), 25 deletions(-) diff --git a/go.mod b/go.mod index 7411214..aab8f7f 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,7 @@ module gitlab.crans.org/nounous/ghostream go 1.15 require ( + github.com/3d0c/gmf v0.0.0-20200614092945-e58d8d5a6035 github.com/go-ldap/ldap/v3 v3.2.3 github.com/gorilla/websocket v1.4.0 github.com/haivision/srtgo v0.0.0-20201025191851-67964e8f497a diff --git a/go.sum b/go.sum index 6628b7c..2374a19 100644 --- a/go.sum +++ b/go.sum @@ -7,6 +7,8 @@ dmitri.shuralyov.com/html/belt v0.0.0-20180602232347-f7d459c86be0/go.mod h1:JLBr dmitri.shuralyov.com/service/change v0.0.0-20181023043359-a85b471d5412/go.mod h1:a1inKt/atXimZ4Mv927x+r7UpyzRUf4emIoiiSC2TN4= dmitri.shuralyov.com/state v0.0.0-20180228185332-28bcc343414c/go.mod h1:0PRwlb0D6DFvNNtx+9ybjezNCa8XF0xaYcETyp6rHWU= git.apache.org/thrift.git v0.0.0-20180902110319-2566ecd5d999/go.mod h1:fPE2ZNJGynbRyZ4dJvy6G277gSllfV2HJqblrnkyeyg= +github.com/3d0c/gmf v0.0.0-20200614092945-e58d8d5a6035 h1:QZb1aMKxiYdGGieyIDmXuw9I9YcGWGViTrpQ6vcZX7Q= +github.com/3d0c/gmf v0.0.0-20200614092945-e58d8d5a6035/go.mod h1:0QMRcUq2JsDECeAq7bj4h79k7XbhtTsrPUQf6G7qfPs= github.com/Azure/go-ntlmssp v0.0.0-20200615164410-66371956d46c h1:/IBSNwUN8+eKzUzbJPqhK839ygXJ82sde8x3ogr6R28= github.com/Azure/go-ntlmssp v0.0.0-20200615164410-66371956d46c/go.mod h1:chxPXzSsl7ZWRAuOIE23GDNzjWuZquvFlgA8xmpunjU= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= diff --git a/stream/webrtc/ingest.go b/stream/webrtc/ingest.go index ff85f90..8a3cb95 100644 --- a/stream/webrtc/ingest.go +++ b/stream/webrtc/ingest.go @@ -3,17 +3,20 @@ package webrtc import ( "bufio" + "bytes" "fmt" - "github.com/pion/rtp" - "github.com/pion/webrtc/v3" - "github.com/pion/webrtc/v3/pkg/media" - "github.com/pion/webrtc/v3/pkg/media/h264reader" - "gitlab.crans.org/nounous/ghostream/messaging" "io" "log" "math/rand" "net" "os/exec" + + "github.com/3d0c/gmf" + "github.com/pion/rtp" + "github.com/pion/webrtc/v3" + "github.com/pion/webrtc/v3/pkg/media" + "github.com/pion/webrtc/v3/pkg/media/h264reader" + "gitlab.crans.org/nounous/ghostream/messaging" ) func ingest(name string, q *messaging.Quality) { @@ -21,26 +24,98 @@ func ingest(name string, q *messaging.Quality) { videoInput := make(chan []byte, 1024) q.Register(videoInput) - // FIXME Mux into RTP without having multiple UDP listeners - firstPort := int(rand.Int31n(63535)) + 2000 + inputCtx := gmf.NewCtx() + avioInputCtx, _ := gmf.NewAVIOContext(inputCtx, &gmf.AVIOHandlers{ReadPacket: func() ([]byte, int) { + data := <-videoInput + return data, len(data) + }}) + log.Println("Open input") + inputCtx.SetPb(avioInputCtx).OpenInput("") + log.Println("Opened") + defer inputCtx.CloseInput() + defer avioInputCtx.Release() - // Open UDP listener for RTP Packets - audioListener, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: firstPort}) - if err != nil { - log.Printf("Faited to open UDP listener %s", err) - return + if audioTracks[name] == nil { + audioTracks[name] = make([]*webrtc.Track, 0) } - // Start ffmpag to convert videoInput to audio UDP - ffmpeg, ffmpegOut, err := startFFmpeg(videoInput, firstPort) + port := rand.Int()%64355 + 2000 + audioListener, _ := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: port}) + + b := bytes.Buffer{} + videoOutputCtx, _ := gmf.NewOutputCtxWithFormatName("/dev/null", "h264") + avioOutputCtx, _ := gmf.NewAVIOContext(videoOutputCtx, &gmf.AVIOHandlers{WritePacket: func(data []byte) int { + n, _ := b.Write(data) + return n + }}) + + videoOutputCtx.SetPb(avioOutputCtx) + defer videoOutputCtx.CloseOutput() + defer avioOutputCtx.Release() + + audioOutputCtx, _ := gmf.NewOutputCtxWithFormatName(fmt.Sprintf("rtp://127.0.0.1:%d", port), "rtp") + defer audioOutputCtx.CloseOutput() + + log.Printf("%d streams", inputCtx.StreamsCnt()) + + c, err := gmf.FindEncoder("libopus") if err != nil { - log.Printf("Error while starting ffmpeg: %s", err) - return + log.Printf("Error while searching opus codec: %s", err) + } + + audioStream, _ := inputCtx.GetBestStream(gmf.AVMEDIA_TYPE_AUDIO) + ctx := gmf.NewCodecCtx(c, []*gmf.Option{ + {Key: "time_base", Val: audioStream.CodecCtx().TimeBase().AVR()}, + {Key: "ar", Val: audioStream.CodecCtx().SampleRate()}, + {Key: "ac", Val: audioStream.CodecCtx().Channels()}, + }) + par := gmf.NewCodecParameters() + _ = par.FromContext(audioStream.CodecCtx()) + defer par.Free() + _, _ = audioOutputCtx.AddStreamWithCodeCtx(ctx) + //c, err = gmf.FindEncoder("libx264") + + videoStream, _ := inputCtx.GetBestStream(gmf.AVMEDIA_TYPE_VIDEO) + c, err = gmf.FindEncoder("libx264") + if err != nil { + log.Printf("Error while searching x264 codec: %s", err) + } + + ctx = gmf.NewCodecCtx(c, []*gmf.Option{ + {Key: "time_base", Val: gmf.AVR{Num: 1, Den: 25}}, + {Key: "pixel_format", Val: gmf.AV_PIX_FMT_YUV420P}, + // Save original + {Key: "video_size", Val: videoStream.CodecCtx().GetVideoSize()}, + {Key: "b", Val: 500000}, + }) + par = gmf.NewCodecParameters() + _ = par.FromContext(videoStream.CodecCtx()) + defer par.Free() + _, _ = videoOutputCtx.AddStreamWithCodeCtx(ctx) + + for i := 0; i < inputCtx.StreamsCnt(); i++ { + srcStream, err := inputCtx.GetStream(i) + if err != nil { + log.Println("GetStream error") + continue + } + + log.Println(srcStream.CodecCtx()) + } + videoOutputCtx.Dump() + audioOutputCtx.Dump() + + if err := videoOutputCtx.WriteHeader(); err != nil { + log.Printf("Unable to write video header: %s", err) + } + + if err := audioOutputCtx.WriteHeader(); err != nil { + log.Printf("Unable to write audio header: %s", err) } // Receive video go func() { - h264, _ := h264reader.NewReader(*ffmpegOut) + h264, _ := h264reader.NewReader(&b) var spsAndPpsCache []byte for { @@ -67,6 +142,7 @@ func ingest(name string, q *messaging.Quality) { nal.Data = append(spsAndPpsCache, nal.Data...) spsAndPpsCache = []byte{} } + log.Println(len(nal.Data)) for _, videoTrack := range videoTracks[name] { if h264Err = videoTrack.WriteSample(media.Sample{Data: nal.Data, Samples: 90000}); h264Err != nil { @@ -108,16 +184,31 @@ func ingest(name string, q *messaging.Quality) { } }() - // Wait for stopped ffmpeg - if err = ffmpeg.Wait(); err != nil { - log.Printf("Faited to wait for ffmpeg: %s", err) + for packet := range inputCtx.GetNewPackets() { + if packet.StreamIndex() == 0 { + if err := videoOutputCtx.WritePacketNoBuffer(packet); err != nil { + log.Printf("Error while writing packet: %s", err) + } + } else if packet.StreamIndex() == 1 { + packet = packet.Clone() + packet.SetStreamIndex(0) + if err := audioOutputCtx.WritePacketNoBuffer(packet); err != nil { + log.Printf("Error while writing packet: %s", err) + } + } + packet.Free() } - // Close UDP listener - if err = audioListener.Close(); err != nil { - log.Printf("Faited to close UDP listener: %s", err) - } - q.Unregister(videoInput) + // Wait for stopped ffmpeg + /* if err = ffmpeg.Wait(); err != nil { + log.Printf("Faited to wait for ffmpeg: %s", err) + } + + // Close UDP listener + if err = audioListener.Close(); err != nil { + log.Printf("Faited to close UDP listener: %s", err) + } + q.Unregister(videoInput)*/ } func startFFmpeg(in <-chan []byte, listeningPort int) (ffmpeg *exec.Cmd, stdout *io.ReadCloser, err error) {