Better error handling in stream forwarder

This commit is contained in:
Alexandre Iooss 2020-09-30 15:28:19 +02:00
parent 2550d75c57
commit 8883c878bf
No known key found for this signature in database
GPG Key ID: 6C79278F3FCDCC02
3 changed files with 35 additions and 26 deletions

View File

@ -96,11 +96,8 @@ func main() {
go web.Serve(remoteSdpChan, localSdpChan, &cfg.Web) go web.Serve(remoteSdpChan, localSdpChan, &cfg.Web)
go webrtc.Serve(remoteSdpChan, localSdpChan, &cfg.WebRTC) go webrtc.Serve(remoteSdpChan, localSdpChan, &cfg.WebRTC)
// Init stream forwarding // Configure stream forwarding
err = forwarding.New(&cfg.Forwarding) forwarding.New(cfg.Forwarding)
if err != nil {
log.Fatalln("Failed to init stream forwarding:", err)
}
// Wait for routines // Wait for routines
select {} select {}

View File

@ -12,56 +12,59 @@ import (
type Options map[string][]string type Options map[string][]string
var ( var (
options Options cfg Options
ffmpegInstances = make(map[string]*exec.Cmd) ffmpegInstances = make(map[string]*exec.Cmd)
ffmpegInputStreams = make(map[string]*io.WriteCloser) ffmpegInputStreams = make(map[string]*io.WriteCloser)
) )
// New Load configuration // New Load configuration
func New(cfg *Options) error { func New(c Options) {
options = *cfg cfg = c
return nil log.Printf("Stream forwarding initialized")
} }
// RegisterStream Declare a new open stream and create ffmpeg instances // RegisterStream Declare a new open stream and create ffmpeg instances
func RegisterStream(name string) { func RegisterStream(name string) error {
if len(options[name]) == 0 { streams, exist := cfg[name]
return if !exist || len(streams) == 0 {
// Nothing to do, not configured
return nil
} }
// Launch FFMPEG instance
params := []string{"-re", "-i", "pipe:0"} params := []string{"-re", "-i", "pipe:0"}
for _, stream := range options[name] { 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)
} }
// Launch FFMPEG instance
ffmpeg := exec.Command("ffmpeg", params...) ffmpeg := exec.Command("ffmpeg", params...)
// Open pipes // Open pipes
input, err := ffmpeg.StdinPipe() input, err := ffmpeg.StdinPipe()
if err != nil { if err != nil {
panic(err) return err
} }
output, err := ffmpeg.StdoutPipe() output, err := ffmpeg.StdoutPipe()
if err != nil { if err != nil {
panic(err) return err
} }
errOutput, err := ffmpeg.StderrPipe() errOutput, err := ffmpeg.StderrPipe()
if err != nil { if err != nil {
panic(err) return err
} }
ffmpegInstances[name] = ffmpeg ffmpegInstances[name] = ffmpeg
ffmpegInputStreams[name] = &input ffmpegInputStreams[name] = &input
// Start FFMpeg
if err := ffmpeg.Start(); err != nil { if err := ffmpeg.Start(); err != nil {
panic(err) return err
} }
// Log ffmpeg output // Log ffmpeg output
go func() { go func() {
scanner := bufio.NewScanner(output) scanner := bufio.NewScanner(output)
for scanner.Scan() { for scanner.Scan() {
log.Println("[FFMPEG " + name + "] " + scanner.Text()) log.Printf("[FFMPEG %s] %s", name, scanner.Text())
} }
}() }()
@ -69,12 +72,14 @@ func RegisterStream(name string) {
go func() { go func() {
scanner := bufio.NewScanner(errOutput) scanner := bufio.NewScanner(errOutput)
for scanner.Scan() { for scanner.Scan() {
log.Println("[FFMPEG ERROR " + name + "] " + scanner.Text()) log.Printf("[FFMPEG ERR %s] %s", name, scanner.Text())
} }
}() }()
return nil
} }
// SendPacket When a SRT packet is received, transmit it to all FFMPEG instances related to the stream key // SendPacket forward data to all FFMpeg instances related to the stream name
func SendPacket(name string, data []byte) { func SendPacket(name string, data []byte) {
stdin := ffmpegInputStreams[name] stdin := ffmpegInputStreams[name]
_, err := (*stdin).Write(data) _, err := (*stdin).Write(data)
@ -84,11 +89,12 @@ func SendPacket(name string, data []byte) {
} }
// CloseConnection When the stream is ended, close FFMPEG instances // CloseConnection When the stream is ended, close FFMPEG instances
func CloseConnection(name string) { func CloseConnection(name string) error {
ffmpeg := ffmpegInstances[name] ffmpeg := ffmpegInstances[name]
if err := ffmpeg.Process.Kill(); err != nil { if err := ffmpeg.Process.Kill(); err != nil {
panic(err) return err
} }
delete(ffmpegInstances, name) delete(ffmpegInstances, name)
delete(ffmpegInputStreams, name) delete(ffmpegInputStreams, name)
return nil
} }

View File

@ -57,21 +57,23 @@ func Serve(cfg *Options) {
buff := make([]byte, 2048) buff := make([]byte, 2048)
// Setup stream forwarding // Setup stream forwarding
forwarding.RegisterStream("demo") // FIXME Replace with real stream key // FIXME: demo should be replaced by stream name
if err := forwarding.RegisterStream("demo"); err != nil {
log.Println("Error occurred during forward stream init:", err)
break
}
// Read RTP packets forever and send them to the WebRTC Client // Read RTP packets forever and send them to the WebRTC Client
for { for {
n, err := s.Read(buff, 10000) n, err := s.Read(buff, 10000)
if err != nil { if err != nil {
log.Println("Error occured while reading SRT socket:", err) log.Println("Error occured while reading SRT socket:", err)
forwarding.CloseConnection("demo")
break break
} }
if n == 0 { if n == 0 {
// End of stream // End of stream
log.Printf("Received no bytes, stopping stream.") log.Printf("Received no bytes, stopping stream.")
forwarding.CloseConnection("demo")
break break
} }
@ -84,5 +86,9 @@ func Serve(cfg *Options) {
// See https://github.com/ebml-go/webm/blob/master/reader.go // See https://github.com/ebml-go/webm/blob/master/reader.go
//err := videoTrack.WriteSample(media.Sample{Data: data, Samples: uint32(sampleCount)}) //err := videoTrack.WriteSample(media.Sample{Data: data, Samples: uint32(sampleCount)})
} }
if err := forwarding.CloseConnection("demo"); err != nil {
log.Printf("Failed to close forward stream: %s", err)
}
} }
} }