Separate the WebRTC stream subroutine in a dedicated subroutine

stream/webrtc/ingest.go

@@ -15,16 +15,49 @@ import (
 	"gitlab.crans.org/nounous/ghostream/stream/srt"
 )
 
+var (
+	ffmpeg      = make(map[string]*exec.Cmd)
+	ffmpegInput = make(map[string]io.WriteCloser)
+)
+
 func ingestFrom(inputChannel chan srt.Packet) {
 	// FIXME Clean code
-	var ffmpeg *exec.Cmd
-	var ffmpegInput io.WriteCloser
 
 	for {
 		var err error = nil
 		srtPacket := <-inputChannel
+		log.Println(len(inputChannel))
 		switch srtPacket.PacketType {
 		case "register":
+			go registerStream(&srtPacket)
+			break
+		case "sendData":
+			if _, ok := ffmpegInput[srtPacket.StreamName]; !ok {
+				break
+			}
+			// FIXME send to stream srtPacket.StreamName
+			if _, err := ffmpegInput[srtPacket.StreamName].Write(srtPacket.Data); err != nil {
+				log.Printf("Failed to write data to ffmpeg input: %s", err)
+			}
+			break
+		case "close":
+			log.Printf("WebRTC CloseConnection %s", srtPacket.StreamName)
+			_ = ffmpeg[srtPacket.StreamName].Process.Kill()
+			_ = ffmpegInput[srtPacket.StreamName].Close()
+			delete(ffmpeg, srtPacket.StreamName)
+			delete(ffmpegInput, srtPacket.StreamName)
+			break
+		default:
+			log.Println("Unknown SRT srtPacket type:", srtPacket.PacketType)
+			break
+		}
+		if err != nil {
+			log.Printf("Error occured while receiving SRT srtPacket of type %s: %s", srtPacket.PacketType, err)
+		}
+	}
+}
+
+func registerStream(srtPacket *srt.Packet) {
 	log.Printf("WebRTC RegisterStream %s", srtPacket.StreamName)
 
 	// Open a UDP Listener for RTP Packets on port 5004
@@ -38,16 +71,17 @@ func ingestFrom(inputChannel chan srt.Packet) {
 		log.Printf("Faited to open UDP listener %s", err)
 		return
 	}
-			defer func() {
+	// FIXME Close UDP listeners at the end of the stream, not the end of the routine
+	/*	defer func() {
 		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)
 		}
-			}()
+	}() */
 
-			ffmpegArgs := []string{"-hide_banner", "-loglevel", "error", "-re", "-i", "pipe:0",
+	ffmpegArgs := []string{"-re", "-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",
@@ -57,32 +91,34 @@ func ingestFrom(inputChannel chan srt.Packet) {
 
 	// Export stream to ascii art
 	if telnet.Cfg.Enabled {
+		bitrate := fmt.Sprintf("%dk", telnet.Cfg.Width*telnet.Cfg.Height/telnet.Cfg.Delay)
 		ffmpegArgs = append(ffmpegArgs,
-					"-an", "-f", "rawvideo", "-vf", fmt.Sprintf("scale=%dx%d", telnet.Cfg.Width, telnet.Cfg.Height), "-pix_fmt", "gray", "pipe:1")
+			"-an", "-vf", fmt.Sprintf("scale=%dx%d", telnet.Cfg.Width, telnet.Cfg.Height),
+			"-b:v", bitrate, "-minrate", bitrate, "-maxrate", bitrate, "-bufsize", bitrate, "-q", "42", "-pix_fmt", "gray", "-f", "rawvideo", "pipe:1")
 	}
 
-			ffmpeg = exec.Command("ffmpeg", ffmpegArgs...)
+	ffmpeg[srtPacket.StreamName] = exec.Command("ffmpeg", ffmpegArgs...)
 
-			input, err := ffmpeg.StdinPipe()
+	input, err := ffmpeg[srtPacket.StreamName].StdinPipe()
 	if err != nil {
 		panic(err)
 	}
-			ffmpegInput = input
+	ffmpegInput[srtPacket.StreamName] = input
-			errOutput, err := ffmpeg.StderrPipe()
+	errOutput, err := ffmpeg[srtPacket.StreamName].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()
+		output, err := ffmpeg[srtPacket.StreamName].StdoutPipe()
 		if err != nil {
 			panic(err)
 		}
 		go telnet.StartASCIIArtStream(srtPacket.StreamName, output)
 	}
 
-			if err := ffmpeg.Start(); err != nil {
+	if err := ffmpeg[srtPacket.StreamName].Start(); err != nil {
 		panic(err)
 	}
 
@@ -156,22 +192,4 @@ func ingestFrom(inputChannel chan srt.Packet) {
 			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 {
-				log.Printf("Failed to write data to ffmpeg input: %s", err)
-			}
-			break
-		case "close":
-			log.Printf("WebRTC CloseConnection %s", srtPacket.StreamName)
-			break
-		default:
-			log.Println("Unknown SRT srtPacket type:", srtPacket.PacketType)
-			break
-		}
-		if err != nil {
-			log.Printf("Error occured while receiving SRT srtPacket of type %s: %s", srtPacket.PacketType, err)
-		}
-	}
 }