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