diff --git a/stream/forwarding/forwarding.go b/stream/forwarding/forwarding.go index a1d4f76..94a6d78 100644 --- a/stream/forwarding/forwarding.go +++ b/stream/forwarding/forwarding.go @@ -13,27 +13,24 @@ import ( // For each stream name, user can provide several URL to forward stream to type Options map[string][]string -var ( - ffmpegInstances = make(map[string]*exec.Cmd) - ffmpegInputStreams = make(map[string]*io.WriteCloser) -) - // Serve handles incoming packets from SRT and forward them to other external services func Serve(inputChannel chan srt.Packet, cfg Options) { log.Printf("Stream forwarding initialized") + ffmpegInstances := make(map[string]*exec.Cmd) + ffmpegInputStreams := make(map[string]*io.WriteCloser) for { var err error = nil // Wait for packets packet := <-inputChannel switch packet.PacketType { case "register": - err = RegisterStream(packet.StreamName, cfg) + err = registerStream(packet.StreamName, ffmpegInstances, ffmpegInputStreams, cfg) break case "sendData": - err = SendPacket(packet.StreamName, packet.Data) + err = sendPacket(packet.StreamName, ffmpegInputStreams, packet.Data) break case "close": - err = CloseConnection(packet.StreamName) + err = close(packet.StreamName, ffmpegInstances, ffmpegInputStreams) break default: log.Println("Unknown SRT packet type:", packet.PacketType) @@ -45,8 +42,8 @@ func Serve(inputChannel chan srt.Packet, cfg Options) { } } -// RegisterStream Declare a new open stream and create ffmpeg instances -func RegisterStream(name string, cfg Options) error { +// 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 @@ -54,7 +51,7 @@ func RegisterStream(name string, cfg Options) error { } // Launch FFMPEG instance - params := []string{"-re", "-i", "pipe:0"} + params := []string{"-hide_banner", "-loglevel", "error", "-re", "-i", "pipe:0"} for _, stream := range streams { params = append(params, "-f", "flv", "-preset", "ultrafast", "-tune", "zerolatency", "-c", "copy", stream) @@ -101,8 +98,8 @@ func RegisterStream(name string, cfg Options) error { return nil } -// SendPacket forward data to all FFMpeg instances related to the stream name -func SendPacket(name string, data []byte) error { +// sendPacket forwards data to the ffmpeg instance related to the stream name +func sendPacket(name string, ffmpegInputStreams map[string]*io.WriteCloser, data []byte) error { stdin := ffmpegInputStreams[name] if stdin == nil { // Don't need to forward stream @@ -112,8 +109,8 @@ func SendPacket(name string, data []byte) error { return err } -// CloseConnection When the stream is ended, close FFMPEG instances -func CloseConnection(name string) error { +// 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 diff --git a/stream/forwarding/forwarding_test.go b/stream/forwarding/forwarding_test.go index 1cb2a1a..5ba96ef 100644 --- a/stream/forwarding/forwarding_test.go +++ b/stream/forwarding/forwarding_test.go @@ -27,7 +27,8 @@ func TestForwardStream(t *testing.T) { } // Start virtual RTMP server with ffmpeg - forwardedFfmpeg := exec.Command("ffmpeg", "-y", // allow overwrite /dev/null + forwardedFfmpeg := exec.Command("ffmpeg", "-hide_banner", "-loglevel", "error", + "-y", // allow overwrite /dev/null "-listen", "1", "-i", "rtmp://127.0.0.1:1936/live/app", "-f", "null", "-c", "copy", "/dev/null") forwardingOutput, err := forwardedFfmpeg.StdoutPipe() forwardingErrOutput, err := forwardedFfmpeg.StderrPipe() @@ -63,7 +64,7 @@ func TestForwardStream(t *testing.T) { // Serve SRT Server without authentification backend go srt.Serve(&srt.Options{ListenAddress: ":9712", MaxClients: 2}, nil, forwardingChannel, nil) - ffmpeg := exec.Command("ffmpeg", + ffmpeg := exec.Command("ffmpeg", "-hide_banner", "-loglevel", "error", "-re", "-f", "lavfi", "-i", "testsrc=size=640x480:rate=10", "-f", "flv", "srt://127.0.0.1:9712?streamid=demo:")