ynerant/ghostream
1
0
Fork 0
mirror of https://gitlab.crans.org/nounous/ghostream.git synced 2024-12-23 00:22:19 +00:00
Code Issues Releases Wiki Activity

Remove globals in forwarding package and make ffmpeg quieter

Browse Source
This commit is contained in:
Alexandre Iooss 2020-10-04 20:42:16 +02:00
parent 606542f622
commit 1576e391eb
No known key found for this signature in database
GPG Key ID: 6C79278F3FCDCC02
2 changed files with 15 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
stream/forwarding/forwarding.go
View File

@ -13,27 +13,24 @@ import (
// For each stream name, user can provide several URL to forward stream to // For each stream name, user can provide several URL to forward stream to
type Options map[string][]string 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 // Serve handles incoming packets from SRT and forward them to other external services
func Serve(inputChannel chan srt.Packet, cfg Options) { func Serve(inputChannel chan srt.Packet, cfg Options) {
log.Printf("Stream forwarding initialized") log.Printf("Stream forwarding initialized")
ffmpegInstances := make(map[string]*exec.Cmd)
ffmpegInputStreams := make(map[string]*io.WriteCloser)
for { for {
var err error = nil var err error = nil
// Wait for packets // Wait for packets
packet := <-inputChannel packet := <-inputChannel
switch packet.PacketType { switch packet.PacketType {
case "register": case "register":
err = RegisterStream(packet.StreamName, cfg) err = registerStream(packet.StreamName, ffmpegInstances, ffmpegInputStreams, cfg)
break break
case "sendData": case "sendData":
err = SendPacket(packet.StreamName, packet.Data) err = sendPacket(packet.StreamName, ffmpegInputStreams, packet.Data)
break break
case "close": case "close":
err = CloseConnection(packet.StreamName) err = close(packet.StreamName, ffmpegInstances, ffmpegInputStreams)
break break
default: default:
log.Println("Unknown SRT packet type:", packet.PacketType) 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 // registerStream creates ffmpeg instance associated with newly created stream
func RegisterStream(name string, cfg Options) error { func registerStream(name string, ffmpegInstances map[string]*exec.Cmd, ffmpegInputStreams map[string]*io.WriteCloser, cfg Options) error {
streams, exist := cfg[name] streams, exist := cfg[name]
if !exist || len(streams) == 0 { if !exist || len(streams) == 0 {
// Nothing to do, not configured // Nothing to do, not configured
@ -54,7 +51,7 @@ func RegisterStream(name string, cfg Options) error {
} }
// Launch FFMPEG instance // Launch FFMPEG instance
params := []string{"-re", "-i", "pipe:0"} params := []string{"-hide_banner", "-loglevel", "error", "-re", "-i", "pipe:0"}
for _, stream := range streams { for _, stream := range streams {
params = append(params, "-f", "flv", "-preset", "ultrafast", "-tune", "zerolatency", params = append(params, "-f", "flv", "-preset", "ultrafast", "-tune", "zerolatency",
"-c", "copy", stream) "-c", "copy", stream)
@ -101,8 +98,8 @@ func RegisterStream(name string, cfg Options) error {
return nil return nil
} }
// SendPacket forward data to all FFMpeg instances related to the stream name // sendPacket forwards data to the ffmpeg instance related to the stream name
func SendPacket(name string, data []byte) error { func sendPacket(name string, ffmpegInputStreams map[string]*io.WriteCloser, data []byte) error {
stdin := ffmpegInputStreams[name] stdin := ffmpegInputStreams[name]
if stdin == nil { if stdin == nil {
// Don't need to forward stream // Don't need to forward stream
@ -112,8 +109,8 @@ func SendPacket(name string, data []byte) error {
return err return err
} }
// CloseConnection When the stream is ended, close FFMPEG instances // close ffmpeg instance associated with stream name
func CloseConnection(name string) error { func close(name string, ffmpegInstances map[string]*exec.Cmd, ffmpegInputStreams map[string]*io.WriteCloser) error {
ffmpeg := ffmpegInstances[name] ffmpeg := ffmpegInstances[name]
if ffmpeg == nil { if ffmpeg == nil {
// No stream to close // No stream to close

5
stream/forwarding/forwarding_test.go
View File

@ -27,7 +27,8 @@ func TestForwardStream(t *testing.T) {
} }
// Start virtual RTMP server with ffmpeg // 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") "-listen", "1", "-i", "rtmp://127.0.0.1:1936/live/app", "-f", "null", "-c", "copy", "/dev/null")
forwardingOutput, err := forwardedFfmpeg.StdoutPipe() forwardingOutput, err := forwardedFfmpeg.StdoutPipe()
forwardingErrOutput, err := forwardedFfmpeg.StderrPipe() forwardingErrOutput, err := forwardedFfmpeg.StderrPipe()
@ -63,7 +64,7 @@ func TestForwardStream(t *testing.T) {
// Serve SRT Server without authentification backend // Serve SRT Server without authentification backend
go srt.Serve(&srt.Options{ListenAddress: ":9712", MaxClients: 2}, nil, forwardingChannel, nil) 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", "-re", "-f", "lavfi", "-i", "testsrc=size=640x480:rate=10",
"-f", "flv", "srt://127.0.0.1:9712?streamid=demo:") "-f", "flv", "srt://127.0.0.1:9712?streamid=demo:")