ghostream/stream/webrtc/ingest.go

// Package webrtc provides the backend to simulate a WebRTC client to send stream
package webrtc

import (
	"bufio"
	"log"
	"net"
	"os/exec"

	"github.com/pion/rtp"
	"github.com/pion/webrtc/v3"
	"gitlab.crans.org/nounous/ghostream/messaging"
)

func ingest(name string, q *messaging.Quality) {
	// Register to get stream
	input := make(chan []byte, 1024)
	// FIXME Stream data should already be transcoded
	source, _ := q.Stream.GetQuality("source")
	source.Register(input)

	// FIXME Bad code
	port := 5000
	var tracks map[string][]*webrtc.Track
	qualityName := ""
	switch q.Name {
	case "audio":
		port = 5004
		tracks = audioTracks
		break
	case "source":
		port = 5005
		tracks = videoTracks
		qualityName = "@source"
		break
	case "480p":
		port = 5006
		tracks = videoTracks
		qualityName = "@480p"
		break
	case "360p":
		port = 5007
		tracks = videoTracks
		qualityName = "@360p"
		break
	case "240p":
		port = 5008
		tracks = videoTracks
		qualityName = "@240p"
		break
	}

	// Open a UDP Listener for RTP Packets
	listener, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: port})
	if err != nil {
		log.Printf("Faited to open UDP listener %s", err)
		return
	}

	// Start ffmpag to convert input to video and audio UDP
	ffmpeg, err := startFFmpeg(q, input)
	if err != nil {
		log.Printf("Error while starting ffmpeg: %s", err)
		return
	}

	// Receive stream
	go func() {
		inboundRTPPacket := make([]byte, 1500) // UDP MTU
		for {
			n, _, err := listener.ReadFromUDP(inboundRTPPacket)
			if err != nil {
				log.Printf("Failed to read from UDP: %s", err)
				break
			}
			packet := &rtp.Packet{}
			if err := packet.Unmarshal(inboundRTPPacket[:n]); err != nil {
				log.Printf("Failed to unmarshal RTP srtPacket: %s", err)
				continue
			}

			// Write RTP srtPacket to all tracks
			// Adapt payload and SSRC to match destination
			for _, track := range tracks[name+qualityName] {
				packet.Header.PayloadType = track.PayloadType()
				packet.Header.SSRC = track.SSRC()
				if writeErr := track.WriteRTP(packet); writeErr != nil {
					log.Printf("Failed to write to track: %s", writeErr)
					continue
				}
			}
		}
	}()

	// Wait for stopped ffmpeg
	if err = ffmpeg.Wait(); err != nil {
		log.Printf("Faited to wait for ffmpeg: %s", err)
	}

	// Close UDP listener
	if err = listener.Close(); err != nil {
		log.Printf("Faited to close UDP listener: %s", err)
	}
	q.Unregister(input)
}

func startFFmpeg(q *messaging.Quality, in <-chan []byte) (ffmpeg *exec.Cmd, err error) {
	// FIXME Use transcoders to downscale, then remux in RTP
	ffmpegArgs := []string{"-hide_banner", "-loglevel", "error", "-i", "pipe:0"}
	switch q.Name {
	case "audio":
		ffmpegArgs = append(ffmpegArgs, "-vn", "-c:a", "libopus", "-b:a", "160k",
			"-f", "rtp", "rtp://127.0.0.1:5004")
		break
	case "source":
		ffmpegArgs = append(ffmpegArgs, "-an", "-c:v", "copy",
			"-f", "rtp", "rtp://127.0.0.1:5005")
		break
	case "480p":
		ffmpegArgs = append(ffmpegArgs,
			"-an", "-c:v", "libx264", "-b:v", "1200k", "-maxrate", "2000k", "-bufsize", "3000k",
			"-preset", "ultrafast", "-profile", "main", "-tune", "zerolatency",
			"-vf", "scale=854:480",
			"-f", "rtp", "rtp://127.0.0.1:5006")
		break
	case "360p":
		ffmpegArgs = append(ffmpegArgs,
			"-an", "-c:v", "libx264", "-b:v", "800k", "-maxrate", "1200k", "-bufsize", "1500k",
			"-preset", "ultrafast", "-profile", "main", "-tune", "zerolatency",
			"-vf", "scale=480:360",
			"-f", "rtp", "rtp://127.0.0.1:5007")
		break
	case "240p":
		ffmpegArgs = append(ffmpegArgs,
			"-an", "-c:v", "libx264", "-b:v", "500k", "-maxrate", "800k", "-bufsize", "1000k",
			"-preset", "ultrafast", "-profile", "main", "-tune", "zerolatency",
			"-vf", "scale=360:240",
			"-f", "rtp", "rtp://127.0.0.1:5008")
		break
	}
	ffmpeg = exec.Command("ffmpeg", ffmpegArgs...)

	// Handle errors output
	errOutput, err := ffmpeg.StderrPipe()
	if err != nil {
		return nil, err
	}
	go func() {
		scanner := bufio.NewScanner(errOutput)
		for scanner.Scan() {
			log.Printf("[WEBRTC FFMPEG %s] %s", "demo", scanner.Text())
		}
	}()

	// Handle stream input
	input, err := ffmpeg.StdinPipe()
	if err != nil {
		return nil, err
	}
	go func() {
		for data := range in {
			if _, err := input.Write(data); err != nil {
				log.Printf("Failed to write data to ffmpeg input: %s", err)
			}
		}

		// End of stream
		ffmpeg.Process.Kill()
	}()

	// Start process
	err = ffmpeg.Start()
	return ffmpeg, err
}