ghostream/stream/forwarding/forwarding.go

122 lines
3.1 KiB
Go
Raw Normal View History

2020-10-09 20:36:02 +00:00
// Package forwarding forwards incoming stream to other streaming services
2020-09-30 13:07:36 +00:00
package forwarding
2020-09-29 19:31:53 +00:00
import (
"bufio"
"io"
"log"
"os/exec"
"gitlab.crans.org/nounous/ghostream/stream/srt"
2020-09-29 19:31:53 +00:00
)
2020-09-30 13:07:36 +00:00
// Options to configure the stream forwarding.
// For each stream name, user can provide several URL to forward stream to
type Options map[string][]string
// Serve handles incoming packets from SRT and forward them to other external services
func Serve(inputChannel chan srt.Packet, cfg Options) {
if len(cfg) < 1 {
// No forwarding, ignore
for {
<-inputChannel // Clear input channel
}
}
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, 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 {
2020-10-14 19:36:12 +00:00
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
}
2020-09-29 19:31:53 +00:00
// Launch FFMPEG instance
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)
}
ffmpeg := exec.Command("ffmpeg", params...)
// Open pipes
input, err := ffmpeg.StdinPipe()
if err != nil {
return err
}
errOutput, err := ffmpeg.StderrPipe()
if err != nil {
return err
}
2020-09-30 13:10:13 +00:00
ffmpegInstances[name] = ffmpeg
ffmpegInputStreams[name] = &input
2020-09-29 19:31:53 +00:00
// Start FFMpeg
if err := ffmpeg.Start(); err != nil {
return err
2020-09-29 19:31:53 +00:00
}
2020-10-04 18:47:05 +00:00
// Log standard error output
go func() {
scanner := bufio.NewScanner(errOutput)
for scanner.Scan() {
2020-10-04 18:47:05 +00:00
log.Printf("[FFMPEG %s] %s", name, scanner.Text())
}
}()
return nil
2020-09-29 19:31:53 +00:00
}
// sendPacket forwards data to the ffmpeg instance related to the stream name
func sendPacket(name string, ffmpegInputStreams map[string]*io.WriteCloser, data []byte) error {
2020-09-30 13:10:13 +00:00
stdin := ffmpegInputStreams[name]
2020-09-30 14:53:15 +00:00
if stdin == nil {
// Don't need to forward stream
return nil
2020-09-30 14:53:15 +00:00
}
_, err := (*stdin).Write(data)
return err
2020-09-29 19:31:53 +00:00
}
// close ffmpeg instance associated with stream name
func close(name string, ffmpegInstances map[string]*exec.Cmd, ffmpegInputStreams map[string]*io.WriteCloser) error {
2020-09-30 13:10:13 +00:00
ffmpeg := ffmpegInstances[name]
2020-09-30 14:53:15 +00:00
if ffmpeg == nil {
// No stream to close
return nil
}
if err := ffmpeg.Process.Kill(); err != nil {
return err
2020-09-29 19:31:53 +00:00
}
2020-09-30 13:10:13 +00:00
delete(ffmpegInstances, name)
delete(ffmpegInputStreams, name)
return nil
2020-09-29 19:31:53 +00:00
}