ghostream/stream/webrtc/ingest.go

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

import (
	"bufio"
	"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"
)

func ingest(name string, q *messaging.Quality) {
	// Register to get stream
	videoInput := make(chan []byte, 1024)
	q.Register(videoInput)

	// FIXME Mux into RTP without having multiple UDP listeners
	firstPort := int(rand.Int31n(63535)) + 2000

	// 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
	}

	// Start ffmpeg to convert videoInput to audio UDP
	ffmpeg, ffmpegOut, err := startFFmpeg(videoInput, firstPort)
	if err != nil {
		log.Printf("Error while starting ffmpeg: %s", err)
		return
	}

	// Receive video
	go func() {
		h264, _ := h264reader.NewReader(bufio.NewReader(*ffmpegOut))
		var spsAndPpsCache []byte

		for {
			nal, h264Err := h264.NextNAL()
			if h264Err == io.EOF {
				fmt.Printf("All video frames parsed and sent")
				return
			}
			if h264Err != nil {
				log.Printf("Failed to read from H264: %s", h264Err)
				break
			}

			if videoTracks[name] == nil {
				videoTracks[name] = make([]*webrtc.Track, 0)
			}

			nal.Data = append([]byte{0x00, 0x00, 0x00, 0x01}, nal.Data...)

			if nal.UnitType == h264reader.NalUnitTypeSPS || nal.UnitType == h264reader.NalUnitTypePPS {
				spsAndPpsCache = append(spsAndPpsCache, nal.Data...)
				continue
			} else if nal.UnitType == h264reader.NalUnitTypeCodedSliceIdr {
				nal.Data = append(spsAndPpsCache, nal.Data...)
				spsAndPpsCache = []byte{}
			}

			for _, videoTrack := range videoTracks[name] {
				if h264Err = videoTrack.WriteSample(media.Sample{Data: nal.Data, Samples: 90000}); h264Err != nil {
					panic(h264Err)
				}
			}
		}
	}()

	// Receive audio
	go func() {
		inboundRTPPacket := make([]byte, 1500) // UDP MTU
		for {
			n, _, err := audioListener.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
			}

			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", err)
					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 = 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) {
	ffmpegArgs := []string{"-hide_banner", "-loglevel", "error", "-i", "pipe:0",
		// Vidéo
		"-an", "-c:v", "copy", "-bsf", "h264_mp4toannexb",
		"-f", "h264", "pipe:1",
		// Audio
		"-vn", "-c:a", "libopus", "-b:a", "96k",
		"-f", "rtp", fmt.Sprintf("rtp://127.0.0.1:%d", listeningPort)}
	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("[WEBRTC FFMPEG %s] %s", "demo", scanner.Text())
		}
	}()

	// Handle stream input
	input, err := ffmpeg.StdinPipe()
	if err != nil {
		return nil, 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()
	}()

	output, err := ffmpeg.StdoutPipe()
	if err != nil {
		return nil, nil, err
	}

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