ghostream/stream/webrtc/ingest.go

178 lines
5.3 KiB
Go
Raw Normal View History

2020-10-09 20:36:02 +00:00
// Package webrtc provides the backend to simulate a WebRTC client to send stream
2020-10-05 08:11:30 +00:00
package webrtc
import (
"bufio"
2020-10-12 22:15:23 +00:00
"fmt"
"gitlab.crans.org/nounous/ghostream/stream/telnet"
2020-10-05 08:11:30 +00:00
"io"
"log"
"net"
"os/exec"
"github.com/pion/rtp"
2020-10-09 20:36:02 +00:00
"github.com/pion/webrtc/v3"
2020-10-05 08:11:30 +00:00
"gitlab.crans.org/nounous/ghostream/stream/srt"
)
func ingestFrom(inputChannel chan srt.Packet) {
// FIXME Clean code
var ffmpeg *exec.Cmd
var ffmpegInput io.WriteCloser
2020-10-05 09:38:17 +00:00
2020-10-05 08:11:30 +00:00
for {
var err error = nil
srtPacket := <-inputChannel
switch srtPacket.PacketType {
2020-10-05 08:11:30 +00:00
case "register":
log.Printf("WebRTC RegisterStream %s", srtPacket.StreamName)
2020-10-05 08:11:30 +00:00
// 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 {
2020-10-05 09:38:17 +00:00
log.Printf("Faited to open UDP listener %s", err)
return
2020-10-05 08:11:30 +00:00
}
audioListener, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 5005})
if err != nil {
2020-10-05 09:38:17 +00:00
log.Printf("Faited to open UDP listener %s", err)
return
2020-10-05 08:11:30 +00:00
}
defer func() {
if err = videoListener.Close(); err != nil {
2020-10-05 09:38:17 +00:00
log.Printf("Faited to close UDP listener %s", err)
2020-10-05 08:11:30 +00:00
}
if err = audioListener.Close(); err != nil {
2020-10-05 09:38:17 +00:00
log.Printf("Faited to close UDP listener %s", err)
2020-10-05 08:11:30 +00:00
}
}()
ffmpegArgs := []string{"-hide_banner", "-loglevel", "error", "-re", "-i", "pipe:0",
2020-10-10 15:05:36 +00:00
"-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",
2020-10-05 08:11:30 +00:00
"-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"}
// Export stream to ascii art
if telnet.Cfg.Enabled {
ffmpegArgs = append(ffmpegArgs,
"-an", "-f", "rawvideo", "-vf", fmt.Sprintf("scale=%dx%d", telnet.Cfg.Width, telnet.Cfg.Height), "-pix_fmt", "gray", "pipe:1")
}
ffmpeg = exec.Command("ffmpeg", ffmpegArgs...)
2020-10-05 08:11:30 +00:00
input, err := ffmpeg.StdinPipe()
if err != nil {
panic(err)
}
ffmpegInput = input
errOutput, err := ffmpeg.StderrPipe()
if err != nil {
panic(err)
}
// Receive raw video output and convert it to ASCII art, then forward it TCP
if telnet.Cfg.Enabled {
output, err := ffmpeg.StdoutPipe()
if err != nil {
panic(err)
}
go telnet.StartASCIIArtStream(srtPacket.StreamName, output)
}
2020-10-05 08:11:30 +00:00
if err := ffmpeg.Start(); err != nil {
panic(err)
}
// Receive video
go func() {
for {
inboundRTPPacket := make([]byte, 1500) // UDP MTU
n, _, err := videoListener.ReadFromUDP(inboundRTPPacket)
if err != nil {
2020-10-05 09:38:17 +00:00
log.Printf("Failed to read from UDP: %s", err)
continue
2020-10-05 08:11:30 +00:00
}
packet := &rtp.Packet{}
if err := packet.Unmarshal(inboundRTPPacket[:n]); err != nil {
log.Printf("Failed to unmarshal RTP srtPacket: %s", err)
2020-10-05 09:38:17 +00:00
continue
2020-10-05 08:11:30 +00:00
}
if videoTracks[srtPacket.StreamName] == nil {
videoTracks[srtPacket.StreamName] = make([]*webrtc.Track, 0)
}
// Write RTP srtPacket to all video tracks
// Adapt payload and SSRC to match destination
for _, videoTrack := range videoTracks[srtPacket.StreamName] {
packet.Header.PayloadType = videoTrack.PayloadType()
packet.Header.SSRC = videoTrack.SSRC()
2020-10-05 08:11:30 +00:00
if writeErr := videoTrack.WriteRTP(packet); writeErr != nil {
2020-10-05 09:38:17 +00:00
log.Printf("Failed to write to video track: %s", err)
continue
2020-10-05 08:11:30 +00:00
}
}
}
}()
// Receive audio
go func() {
for {
inboundRTPPacket := make([]byte, 1500) // UDP MTU
n, _, err := audioListener.ReadFromUDP(inboundRTPPacket)
if err != nil {
2020-10-05 09:38:17 +00:00
log.Printf("Failed to read from UDP: %s", err)
continue
2020-10-05 08:11:30 +00:00
}
packet := &rtp.Packet{}
if err := packet.Unmarshal(inboundRTPPacket[:n]); err != nil {
log.Printf("Failed to unmarshal RTP srtPacket: %s", err)
2020-10-05 09:38:17 +00:00
continue
2020-10-05 08:11:30 +00:00
}
2020-10-05 09:08:22 +00:00
if audioTracks[srtPacket.StreamName] == nil {
audioTracks[srtPacket.StreamName] = make([]*webrtc.Track, 0)
}
// Write RTP srtPacket to all audio tracks
2020-10-05 09:08:22 +00:00
// Adapt payload and SSRC to match destination
for _, audioTrack := range audioTracks[srtPacket.StreamName] {
packet.Header.PayloadType = audioTrack.PayloadType()
packet.Header.SSRC = audioTrack.SSRC()
2020-10-05 08:11:30 +00:00
if writeErr := audioTrack.WriteRTP(packet); writeErr != nil {
2020-10-05 09:38:17 +00:00
log.Printf("Failed to write to audio track: %s", err)
continue
2020-10-05 08:11:30 +00:00
}
}
}
}()
go func() {
scanner := bufio.NewScanner(errOutput)
for scanner.Scan() {
log.Printf("[WEBRTC FFMPEG %s] %s", "demo", scanner.Text())
}
}()
break
case "sendData":
// FIXME send to stream srtPacket.StreamName
if _, err := ffmpegInput.Write(srtPacket.Data); err != nil {
2020-10-05 09:38:17 +00:00
log.Printf("Failed to write data to ffmpeg input: %s", err)
2020-10-05 08:11:30 +00:00
}
break
case "close":
log.Printf("WebRTC CloseConnection %s", srtPacket.StreamName)
2020-10-05 08:11:30 +00:00
break
default:
log.Println("Unknown SRT srtPacket type:", srtPacket.PacketType)
2020-10-05 08:11:30 +00:00
break
}
if err != nil {
log.Printf("Error occured while receiving SRT srtPacket of type %s: %s", srtPacket.PacketType, err)
2020-10-05 08:11:30 +00:00
}
}
}