Fix the forwarding package with the new messaging package

stream/forwarding/forwarding.go

@@ -2,7 +2,9 @@
 package forwarding
 
 import (
+	"bufio"
 	"log"
+	"os/exec"
 	"time"
 
 	"gitlab.crans.org/nounous/ghostream/stream"
@@ -40,107 +42,59 @@ func Serve(streams map[string]*stream.Stream, cfg Options) {
 	}
 }
 
+// Start a FFMPEG instance and redirect stream output to forwarded streams
 func forward(st *stream.Stream, fwdCfg []string) {
-	// FIXME
+	output := make(chan []byte, 1024)
-	/*ffmpegInstances := make(map[string]*exec.Cmd)
+	st.Register(output)
-	ffmpegInputStreams := make(map[string]*io.WriteCloser)
-	for {
-		var err error = nil
-		// Wait for packets
-		// FIXME packet := <-inputChannel
-		packet := srt.Packet{
-			Data:       []byte{},
-			PacketType: "nothing",
-			StreamName: "demo",
-		}
-		switch packet.PacketType {
-		case "register":
-			err = registerStream(packet.StreamName, ffmpegInstances, ffmpegInputStreams, cfg)
-			break
-		case "sendData":
-			err = sendPacket(packet.StreamName, ffmpegInputStreams, packet.Data)
-			break
-		case "close":
-			err = close(packet.StreamName, ffmpegInstances, ffmpegInputStreams)
-			break
-		default:
-			log.Println("Unknown SRT packet type:", packet.PacketType)
-			break
-		}
-		if err != nil {
-			log.Printf("Error occurred while receiving SRT packet of type %s: %s", packet.PacketType, err)
-		}
-	}*/
-}
-
-/*
-// registerStream creates ffmpeg instance associated with newly created stream
-func registerStream(name string, ffmpegInstances map[string]*exec.Cmd, ffmpegInputStreams map[string]*io.WriteCloser, cfg Options) error {
-	streams, exist := cfg[name]
-	if !exist || len(streams) == 0 {
-		// Nothing to do, not configured
-		return nil
-	}
 
 	// Launch FFMPEG instance
 	params := []string{"-hide_banner", "-loglevel", "error", "-re", "-i", "pipe:0"}
-	for _, stream := range streams {
+	for _, url := range fwdCfg {
 		params = append(params, "-f", "flv", "-preset", "ultrafast", "-tune", "zerolatency",
-			"-c", "copy", stream)
+			"-c", "copy", url)
 	}
 	ffmpeg := exec.Command("ffmpeg", params...)
 
 	// Open pipes
 	input, err := ffmpeg.StdinPipe()
 	if err != nil {
-		return err
+		log.Printf("Error while opening forwarding ffmpeg input pipe: %s", err)
+		return
 	}
 	errOutput, err := ffmpeg.StderrPipe()
 	if err != nil {
-		return err
+		log.Printf("Error while opening forwarding ffmpeg output pipe: %s", err)
+		return
 	}
-	ffmpegInstances[name] = ffmpeg
-	ffmpegInputStreams[name] = &input
 
 	// Start FFMpeg
 	if err := ffmpeg.Start(); err != nil {
-		return err
+		log.Printf("Error while starting forwarding ffmpeg instance: %s", err)
 	}
 
+	// Kill FFMPEG when stream is ended
+	defer func() {
+		_ = input.Close()
+		_ = errOutput.Close()
+		_ = ffmpeg.Process.Kill()
+		st.Unregister(output)
+	}()
+
 	// Log standard error output
 	go func() {
 		scanner := bufio.NewScanner(errOutput)
 		for scanner.Scan() {
-			log.Printf("[FFMPEG %s] %s", name, scanner.Text())
+			log.Printf("[FORWARDING FFMPEG] %s", scanner.Text())
 		}
 	}()
 
-	return nil
+	// Read stream output and redirect immediately to ffmpeg
-}
+	for data := range output {
+		_, err := input.Write(data)
 
-// sendPacket forwards data to the ffmpeg instance related to the stream name
+		if err != nil {
-func sendPacket(name string, ffmpegInputStreams map[string]*io.WriteCloser, data []byte) error {
+			log.Printf("Error while writing to forwarded stream: %s", err)
-	stdin := ffmpegInputStreams[name]
+			break
-	if stdin == nil {
+		}
-		// Don't need to forward stream
-		return nil
 	}
-	_, err := (*stdin).Write(data)
-	return err
 }
-
-// close ffmpeg instance associated with stream name
-func close(name string, ffmpegInstances map[string]*exec.Cmd, ffmpegInputStreams map[string]*io.WriteCloser) error {
-	ffmpeg := ffmpegInstances[name]
-	if ffmpeg == nil {
-		// No stream to close
-		return nil
-	}
-	if err := ffmpeg.Process.Kill(); err != nil {
-		return err
-	}
-	delete(ffmpegInstances, name)
-	delete(ffmpegInputStreams, name)
-	return nil
-}
-*/