ynerant/ghostream
1
0
Fork 0
mirror of https://gitlab.crans.org/nounous/ghostream.git synced 2025-06-28 17:32:47 +02:00
Code Issues Releases Wiki Activity

Try to use manually FFMPEG bindings to avoid having too much syscalls, not working yet

Browse Source
This commit is contained in:
Yohann D'ANELLO
2020-11-08 00:08:52 +01:00
parent cc40226111
commit d92bb1d465
3 changed files with 119 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

141
stream/webrtc/ingest.go
View File

@ -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) {