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