diff --git a/go.mod b/go.mod index 128eef3..d98b5c6 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,7 @@ module gitlab.crans.org/nounous/ghostream go 1.13 require ( + github.com/3d0c/gmf v0.0.0-20200614092945-e58d8d5a6035 github.com/go-ldap/ldap/v3 v3.2.3 github.com/haivision/srtgo v0.0.0-20200731151239-e00427ae473a github.com/markbates/pkger v0.17.1 diff --git a/go.sum b/go.sum index 00c058c..b01222b 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 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ= diff --git a/stream/webrtc/ingest.go b/stream/webrtc/ingest.go index cf11ce0..ce6e778 100644 --- a/stream/webrtc/ingest.go +++ b/stream/webrtc/ingest.go @@ -2,13 +2,14 @@ package webrtc import ( + "github.com/3d0c/gmf" "github.com/pion/rtp" "github.com/pion/webrtc/v3" + "gitlab.crans.org/nounous/ghostream/stream" "log" + "net" "strings" "time" - - "gitlab.crans.org/nounous/ghostream/stream" ) var ( @@ -52,11 +53,63 @@ func ingest(name string, input *stream.Stream, audio *stream.Stream) { audio.Register(audioInput) activeStream[name] = struct{}{} + inputCtx := gmf.NewCtx() + avioInputCtx, _ := gmf.NewAVIOContext(inputCtx, &gmf.AVIOHandlers{ReadPacket: func() ([]byte, int) { + data := <-audioInput + return data, len(data) + }}) + log.Println("Open input") + inputCtx.SetPb(avioInputCtx).OpenInput("") + log.Println("Opened") + defer inputCtx.CloseInput() + defer avioInputCtx.Release() + + if audioTracks[name] == nil { + audioTracks[name] = make([]*webrtc.Track, 0) + } + + udpListener, _ := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 1234}) + + outputCtx, _ := gmf.NewOutputCtxWithFormatName("rtp://127.0.0.1:1234", "rtp") + avioOutputCtx, _ := gmf.NewAVIOContext(outputCtx, &gmf.AVIOHandlers{WritePacket: func(data []byte) int { + n := len(data) + log.Printf("Read %d bytes", n) + + return n + }}) + // FIXME DON'T RAN AN UDP LISTENER, PLIZ GET DIRECTLY UDP PACKETS, WHY IS IT SO COMPLICATED???? + // outputCtx.SetPb(avioOutputCtx) + defer outputCtx.CloseOutput() + defer avioOutputCtx.Release() + + log.Printf("%d streams", inputCtx.StreamsCnt()) + + for i := 0; i < inputCtx.StreamsCnt(); i++ { + srcStream, err := inputCtx.GetStream(i) + if err != nil { + log.Println("GetStream error") + } + + outputCtx.AddStreamWithCodeCtx(srcStream.CodecCtx()) + } + outputCtx.Dump() + + if err := outputCtx.WriteHeader(); err != nil { + log.Printf("Unable to write RTP header: %s", err) + } + // Receive audio data go func() { + buff := make([]byte, 1500) for { + n, _ := udpListener.Read(buff) + + if n == 0 { + return + } + packet := &rtp.Packet{} - if err := packet.Unmarshal(<-audioInput); err != nil { + if err := packet.Unmarshal(buff[:n]); err != nil { log.Printf("Failed to unmarshal RTP srtPacket: %s", err) continue } @@ -78,6 +131,19 @@ func ingest(name string, input *stream.Stream, audio *stream.Stream) { } }() + first := false + for packet := range inputCtx.GetNewPackets() { + if first { //if read from rtsp ,the first packets is wrong. + if err := outputCtx.WritePacketNoBuffer(packet); err != nil { + log.Printf("Error while writing packet: %s", err) + } + } + + first = true + + packet.Free() + } + select {} // TODO Register to all substreams and make RTP packets. Don't transcode in this package. diff --git a/transcoder/audio/audio.go b/transcoder/audio/audio.go index 108eec8..27fd6e1 100644 --- a/transcoder/audio/audio.go +++ b/transcoder/audio/audio.go @@ -4,9 +4,9 @@ package audio import ( "bufio" "fmt" + "github.com/3d0c/gmf" + "io" "log" - "math/rand" - "net" "os/exec" "strings" "time" @@ -63,15 +63,15 @@ func transcode(input, output *stream.Stream, cfg *Options) { // Start ffmpeg to transcode video to audio videoInput := make(chan []byte, 1024) input.Register(videoInput) - ffmpeg, rawvideo, err := startFFmpeg(videoInput, cfg) + ffmpeg, audio, err := startFFmpeg(videoInput, cfg) if err != nil { log.Printf("Error while starting ffmpeg: %s", err) return } - dataBuff := make([]byte, 1316) // UDP MTU + dataBuff := make([]byte, gmf.IO_BUFFER_SIZE) // UDP MTU for { - n, err := (*rawvideo).Read(dataBuff) + n, err := (*audio).Read(dataBuff) if err != nil { log.Printf("An error occurred while reading input: %s", err) break @@ -86,36 +86,36 @@ func transcode(input, output *stream.Stream, cfg *Options) { // Stop transcode _ = ffmpeg.Process.Kill() - _ = rawvideo.Close() + _ = (*audio).Close() } // Start a ffmpeg instance to convert stream into audio -func startFFmpeg(in <-chan []byte, cfg *Options) (*exec.Cmd, *net.UDPConn, error) { +func startFFmpeg(in <-chan []byte, cfg *Options) (*exec.Cmd, *io.ReadCloser, error) { // TODO in a future release: remove FFMPEG dependency and transcode directly using the libopus API // FIXME It seems impossible to get a RTP Packet from standard output. // We need to find a clean solution, without waiting on UDP listeners. // FIXME We should also not build RTP packets here. - port := 0 - var udpListener *net.UDPConn - var err error - for { - port = rand.Intn(65535) - udpListener, err = net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: port}) - if err != nil { - if strings.Contains(fmt.Sprintf("%s", err), "address already in use") { - continue + /* port := 0 + var udpListener *net.UDPConn + var err error + for { + port = rand.Intn(65535) + udpListener, err = net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: port}) + if err != nil { + if strings.Contains(fmt.Sprintf("%s", err), "address already in use") { + continue + } + return nil, nil, err } - return nil, nil, err - } - break - } + break + }*/ bitrate := fmt.Sprintf("%dk", cfg.Bitrate) // Use copy audio codec, assume for now that libopus is used by the streamer ffmpegArgs := []string{"-hide_banner", "-loglevel", "error", "-i", "pipe:0", - "-vn", "-c:a", "copy", "-b:a", bitrate, "-f", "rtp", fmt.Sprintf("rtp://127.0.0.1:%d", port)} + "-vn", "-c:a", "copy", "-b:a", bitrate, "-f", "mpegts", "pipe:1"} ffmpeg := exec.Command("ffmpeg", ffmpegArgs...) // Handle errors output @@ -130,6 +130,12 @@ func startFFmpeg(in <-chan []byte, cfg *Options) (*exec.Cmd, *net.UDPConn, error } }() + // Handle text output + output, err := ffmpeg.StdoutPipe() + if err != nil { + return nil, nil, err + } + // Handle stream input input, err := ffmpeg.StdinPipe() if err != nil { @@ -143,5 +149,5 @@ func startFFmpeg(in <-chan []byte, cfg *Options) (*exec.Cmd, *net.UDPConn, error // Start process err = ffmpeg.Start() - return ffmpeg, udpListener, err + return ffmpeg, &output, err }