Try to mux into RTP with FFMPEG bindings

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

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=

internal/config/config.go

@@ -2,6 +2,7 @@
 package config
 
 import (
+	"gitlab.crans.org/nounous/ghostream/transcoder/audio"
 	"net"
 
 	"github.com/sherifabdlnaby/configuro"
@@ -59,6 +60,10 @@ func New() *Config {
 			ListenAddress: ":8023",
 		},
 		Transcoder: transcoder.Options{
+			Audio: audio.Options{
+				Enabled: true,
+				Bitrate: 160,
+			},
 			Text: text.Options{
 				Enabled:   false,
 				Width:     80,

stream/webrtc/ingest.go

@@ -2,16 +2,14 @@
 package webrtc
 
 import (
-	"bufio"
-	"log"
-	"net"
-	"os/exec"
-	"strings"
-	"time"
-
+	"github.com/3d0c/gmf"
 	"github.com/pion/rtp"
 	"github.com/pion/webrtc/v3"
 	"gitlab.crans.org/nounous/ghostream/stream"
+	"log"
+	"net"
+	"strings"
+	"time"
 )
 
 var (
@@ -35,7 +33,9 @@ func autoIngest(streams map[string]*stream.Stream) {
 
 			// Start ingestion
 			log.Printf("Starting webrtc for '%s'", name)
-			go ingest(name, st)
+			// FIXME Ensure that the audio stream exist, but that's poop code
+			time.Sleep(time.Second)
+			go ingest(name, st, streams[name+"@audio"])
 		}
 
 		// Regulary pull stream list,
@@ -45,13 +45,110 @@ func autoIngest(streams map[string]*stream.Stream) {
 	}
 }
 
-func ingest(name string, input *stream.Stream) {
+func ingest(name string, input *stream.Stream, audio *stream.Stream) {
 	// Register to get stream
 	videoInput := make(chan []byte, 1024)
 	input.Register(videoInput)
+	audioInput := make(chan []byte, 1024)
+	audio.Register(audioInput)
 	activeStream[name] = struct{}{}
 
-	// Open a UDP Listener for RTP Packets on port 5004
+	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(buff[:n]); err != nil {
+				log.Printf("Failed to unmarshal RTP srtPacket: %s", err)
+				continue
+			}
+
+			if audioTracks[name] == nil {
+				audioTracks[name] = make([]*webrtc.Track, 0)
+			}
+
+			// Write RTP srtPacket to all audio tracks
+			// Adapt payload and SSRC to match destination
+			for _, audioTrack := range audioTracks[name] {
+				packet.Header.PayloadType = audioTrack.PayloadType()
+				packet.Header.SSRC = audioTrack.SSRC()
+				if writeErr := audioTrack.WriteRTP(packet); writeErr != nil {
+					log.Printf("Failed to write to audio track: %s", writeErr)
+					continue
+				}
+			}
+		}
+	}()
+
+	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.
+
+	/*	// Open a UDP Listener for RTP Packets on port 5004
 		videoListener, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 5004})
 		if err != nil {
 			log.Printf("Faited to open UDP listener %s", err)
@@ -145,11 +242,11 @@ func ingest(name string, input *stream.Stream) {
 		}
 		if err = audioListener.Close(); err != nil {
 			log.Printf("Faited to close UDP listener: %s", err)
-	}
+		}*/
 	delete(activeStream, name)
 }
 
-func startFFmpeg(in <-chan []byte) (ffmpeg *exec.Cmd, err error) {
+/* func startFFmpeg(in <-chan []byte) (ffmpeg *exec.Cmd, err error) {
 	ffmpegArgs := []string{"-hide_banner", "-loglevel", "error", "-i", "pipe:0",
 		"-an", "-vcodec", "libvpx", "-crf", "10", "-cpu-used", "5", "-b:v", "6000k", "-maxrate", "8000k", "-bufsize", "12000k", // TODO Change bitrate when changing quality
 		"-qmin", "10", "-qmax", "42", "-threads", "4", "-deadline", "1", "-error-resilient", "1",
@@ -190,4 +287,4 @@ func startFFmpeg(in <-chan []byte) (ffmpeg *exec.Cmd, err error) {
 	// Start process
 	err = ffmpeg.Start()
 	return ffmpeg, err
-}
+} */

transcoder/audio/audio.go

@@ -0,0 +1,153 @@
+// Package audio transcode a stream to filter the audio
+package audio
+
+import (
+	"bufio"
+	"fmt"
+	"github.com/3d0c/gmf"
+	"io"
+	"log"
+	"os/exec"
+	"strings"
+	"time"
+
+	"gitlab.crans.org/nounous/ghostream/stream"
+)
+
+// Options holds audio package configuration
+type Options struct {
+	Enabled bool
+	Bitrate int
+}
+
+// Init text transcoder
+func Init(streams map[string]*stream.Stream, cfg *Options) {
+	if !cfg.Enabled {
+		// Audio transcode is not enabled, ignore
+		return
+	}
+
+	// Regulary check existing streams
+	for {
+		for sourceName, sourceStream := range streams {
+			if strings.Contains(sourceName, "@") {
+				// Not a source stream, pass
+				continue
+			}
+
+			// Check that the transcoded stream does not already exist
+			name := sourceName + "@audio"
+			_, ok := streams[name]
+			if ok {
+				// Stream is already transcoded
+				continue
+			}
+
+			// Start conversion
+			log.Printf("Starting audio transcode '%s'", name)
+			st := stream.New()
+			streams[name] = st
+
+			go transcode(sourceStream, st, cfg)
+		}
+
+		// Regulary pull stream list,
+		// it may be better to tweak the messaging system
+		// to get an event on a new stream.
+		time.Sleep(time.Second)
+	}
+}
+
+// Extract audio from stream
+func transcode(input, output *stream.Stream, cfg *Options) {
+	// Start ffmpeg to transcode video to audio
+	videoInput := make(chan []byte, 1024)
+	input.Register(videoInput)
+	ffmpeg, audio, err := startFFmpeg(videoInput, cfg)
+	if err != nil {
+		log.Printf("Error while starting ffmpeg: %s", err)
+		return
+	}
+
+	dataBuff := make([]byte, gmf.IO_BUFFER_SIZE) // UDP MTU
+	for {
+		n, err := (*audio).Read(dataBuff)
+		if err != nil {
+			log.Printf("An error occurred while reading input: %s", err)
+			break
+		}
+		if n == 0 {
+			// Stream is finished
+			break
+		}
+
+		output.Broadcast <- dataBuff[:n]
+	}
+
+	// Stop transcode
+	_ = ffmpeg.Process.Kill()
+	_ = (*audio).Close()
+}
+
+// Start a ffmpeg instance to convert stream into audio
+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
+				}
+				return nil, nil, err
+			}
+			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", "mpegts", "pipe:1"}
+	ffmpeg := exec.Command("ffmpeg", ffmpegArgs...)
+
+	// Handle errors output
+	errOutput, err := ffmpeg.StderrPipe()
+	if err != nil {
+		return nil, nil, err
+	}
+	go func() {
+		scanner := bufio.NewScanner(errOutput)
+		for scanner.Scan() {
+			log.Printf("[AUDIO FFMPEG %s] %s", "demo", scanner.Text())
+		}
+	}()
+
+	// Handle text output
+	output, err := ffmpeg.StdoutPipe()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	// Handle stream input
+	input, err := ffmpeg.StdinPipe()
+	if err != nil {
+		return nil, nil, err
+	}
+	go func() {
+		for data := range in {
+			_, _ = input.Write(data)
+		}
+	}()
+
+	// Start process
+	err = ffmpeg.Start()
+	return ffmpeg, &output, err
+}

transcoder/audio/audio_test.go

@@ -0,0 +1 @@
+package audio

transcoder/transcoder.go

@@ -3,15 +3,18 @@ package transcoder
 
 import (
 	"gitlab.crans.org/nounous/ghostream/stream"
+	"gitlab.crans.org/nounous/ghostream/transcoder/audio"
 	"gitlab.crans.org/nounous/ghostream/transcoder/text"
 )
 
 // Options holds text package configuration
 type Options struct {
 	Text  text.Options
+	Audio audio.Options
 }
 
 // Init all transcoders
 func Init(streams map[string]*stream.Stream, cfg *Options) {
 	go text.Init(streams, &cfg.Text)
+	go audio.Init(streams, &cfg.Audio)
 }