ghostream/stream/webrtc/ingest.go

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

import (
	"github.com/3d0c/gmf"
	"github.com/pion/rtp"
	"github.com/pion/webrtc/v3"
	"gitlab.crans.org/nounous/ghostream/stream"
	"log"
	"net"
	"strings"
	"time"
)

var (
	activeStream map[string]struct{}
)

func autoIngest(streams map[string]*stream.Stream) {
	// Regulary check existing streams
	activeStream = make(map[string]struct{})
	for {
		for name, st := range streams {
			if strings.Contains(name, "@") {
				// Not a source stream, pass
				continue
			}

			if _, ok := activeStream[name]; ok {
				// Stream is already ingested
				continue
			}

			// Start ingestion
			log.Printf("Starting webrtc for '%s'", name)
			// 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,
		// it may be better to tweak the messaging system
		// to get an event on a new stream.
		time.Sleep(time.Second)
	}
}

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

	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)
			return
		}
		audioListener, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 5005})
		if err != nil {
			log.Printf("Faited to open UDP listener %s", err)
			return
		}

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

		// Receive video
		go func() {
			inboundRTPPacket := make([]byte, 1500) // UDP MTU
			for {
				n, _, err := videoListener.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 videoTracks[name] == nil {
					videoTracks[name] = make([]*webrtc.Track, 0)
				}

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

		// 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 listeners
		if err = videoListener.Close(); err != nil {
			log.Printf("Faited to close UDP listener: %s", err)
		}
		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) {
	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",
		"-auto-alt-ref", "1",
		"-f", "rtp", "rtp://127.0.0.1:5004",
		"-vn", "-acodec", "libopus", "-cpu-used", "5", "-deadline", "1", "-qmin", "10", "-qmax", "42", "-error-resilient", "1", "-auto-alt-ref", "1",
		"-f", "rtp", "rtp://127.0.0.1:5005"}
	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
} */