From a0d27df8c6a4e7fb62be22d4e69c6e44753fce9f Mon Sep 17 00:00:00 2001 From: Gabe Kangas Date: Sat, 11 Jul 2020 16:04:02 -0700 Subject: [PATCH] Re-work RTMP for #34 --- core/rtmp/rtmp.go | 115 +++++++++++++++++++++++----------------------- go.mod | 2 +- go.sum | 6 ++- utils/utils.go | 2 +- 4 files changed, 64 insertions(+), 61 deletions(-) diff --git a/core/rtmp/rtmp.go b/core/rtmp/rtmp.go index 1a6b3d169..b2b9bb1d3 100644 --- a/core/rtmp/rtmp.go +++ b/core/rtmp/rtmp.go @@ -1,24 +1,25 @@ package rtmp import ( + "fmt" "io" "net" "os" "strings" "syscall" "time" + "unsafe" - "github.com/nareix/joy4/av/avutil" - "github.com/nareix/joy4/format/ts" + "github.com/nareix/joy5/av" + "github.com/nareix/joy5/format/flv" log "github.com/sirupsen/logrus" "github.com/gabek/owncast/config" "github.com/gabek/owncast/core" "github.com/gabek/owncast/core/ffmpeg" "github.com/gabek/owncast/utils" - - "github.com/nareix/joy4/format" - "github.com/nareix/joy4/format/rtmp" + "github.com/nareix/joy5/codec/h264" + "github.com/nareix/joy5/format/rtmp" ) var ( @@ -26,36 +27,53 @@ var ( _isConnected = false ) -func init() { - format.RegisterAll() -} +var _transcoder ffmpeg.Transcoder +var _pipe *os.File //Start starts the rtmp service, listening on port 1935 func Start() { port := 1935 - server := &rtmp.Server{} + s := rtmp.NewServer() + var lis net.Listener + var error error + if lis, error = net.Listen("tcp", fmt.Sprintf(":%d", port)); error != nil { + return + } - server.HandlePublish = handlePublish + s.LogEvent = func(c *rtmp.Conn, nc net.Conn, e int) { + es := rtmp.EventString[e] + log.Traceln(unsafe.Pointer(c), nc.LocalAddr(), nc.RemoteAddr(), es) + } + + s.HandleConn = HandleConn - error := server.ListenAndServe() if error != nil { log.Panicln(error) } log.Infof("RTMP server is listening for incoming stream on port: %d", port) + + for { + nc, err := lis.Accept() + if err != nil { + time.Sleep(time.Second) + continue + } + go s.HandleNetConn(nc) + } } -func handlePublish(conn *rtmp.Conn) { +func HandleConn(c *rtmp.Conn, nc net.Conn) { if _isConnected { log.Errorln("stream already running; can not overtake an existing stream") - conn.Close() + nc.Close() return } - streamingKeyComponents := strings.Split(conn.URL.Path, "/") + streamingKeyComponents := strings.Split(c.URL.Path, "/") streamingKey := streamingKeyComponents[len(streamingKeyComponents)-1] if streamingKey != config.Config.VideoSettings.StreamingKey { log.Errorln("invalid streaming key; rejecting incoming stream") - conn.Close() + nc.Close() return } @@ -63,67 +81,50 @@ func handlePublish(conn *rtmp.Conn) { pipePath := utils.GetTemporaryPipePath() syscall.Mkfifo(pipePath, 0666) - transcoder := ffmpeg.NewTranscoder() - go transcoder.Start() + + _transcoder = ffmpeg.NewTranscoder() + go _transcoder.Start() _isConnected = true core.SetStreamAsConnected() - f, err := os.OpenFile(pipePath, os.O_WRONLY, os.ModeNamedPipe) + f, err := os.OpenFile(pipePath, os.O_RDWR, os.ModeNamedPipe) + _pipe = f + fmt.Println(pipePath) if err != nil { panic(err) } - // Is this too fast? Are there downsides to peeking - // into the stream so frequently? - ticker := time.NewTicker(500 * time.Millisecond) - go func() { - for { - select { - case <-ticker.C: - error := connCheck(conn.NetConn()) - if error == io.EOF { - handleDisconnect(conn) - } + w := flv.NewMuxer(f) + + for { + pkt, err := c.ReadPacket() + if err != nil { + if err == io.EOF { + handleDisconnect(nc) } } - }() - muxer := ts.NewMuxer(f) - avutil.CopyFile(muxer, conn) -} -// Proactively check if the RTMP connection is still active or not. -// Taken from https://stackoverflow.com/a/58664631. -func connCheck(conn net.Conn) error { - var sysErr error = nil - rc, err := conn.(syscall.Conn).SyscallConn() - if err != nil { - return err - } - err = rc.Read(func(fd uintptr) bool { - var buf []byte = []byte{0} - n, _, err := syscall.Recvfrom(int(fd), buf, syscall.MSG_PEEK|syscall.MSG_DONTWAIT) - switch { - case n == 0 && err == nil: - sysErr = io.EOF - case err == syscall.EAGAIN || err == syscall.EWOULDBLOCK: - sysErr = nil - default: - sysErr = err + if pkt.Type == av.H264 { + nalus, _ := h264.SplitNALUs(pkt.Data) + annexb := h264.JoinNALUsAnnexb(nalus) + avcc := h264.JoinNALUsAVCC([][]byte{annexb}) + pkt.Data = avcc + } + + if err := w.WritePacket(pkt); err != nil { + panic(err) } - return true - }) - if err != nil { - return err } - return sysErr } -func handleDisconnect(conn *rtmp.Conn) { +func handleDisconnect(conn net.Conn) { log.Infoln("RTMP disconnected.") conn.Close() + _pipe.Close() _isConnected = false + _transcoder.Stop() core.SetStreamAsDisconnected() } diff --git a/go.mod b/go.mod index cb34d15e5..5813622f7 100644 --- a/go.mod +++ b/go.mod @@ -12,7 +12,7 @@ require ( github.com/libp2p/go-libp2p-peerstore v0.2.6 github.com/mssola/user_agent v0.5.2 github.com/multiformats/go-multiaddr v0.2.2 - github.com/nareix/joy4 v0.0.0-20200507095837-05a4ffbb5369 + github.com/nareix/joy5 v0.0.0-20200710135721-d57196c8d506 github.com/radovskyb/watcher v1.0.7 github.com/sirupsen/logrus v1.6.0 github.com/teris-io/shortid v0.0.0-20171029131806-771a37caa5cf diff --git a/go.sum b/go.sum index 12124bbcd..fad8e48db 100644 --- a/go.sum +++ b/go.sum @@ -803,8 +803,8 @@ github.com/multiformats/go-varint v0.0.2/go.mod h1:3Ls8CIEsrijN6+B7PbrXRPxHRPuXS github.com/multiformats/go-varint v0.0.5 h1:XVZwSo04Cs3j/jS0uAEPpT3JY6DzMcVLLoWOSnCxOjg= github.com/multiformats/go-varint v0.0.5/go.mod h1:3Ls8CIEsrijN6+B7PbrXRPxHRPuXSrVKRY101jdMZYE= github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= -github.com/nareix/joy4 v0.0.0-20200507095837-05a4ffbb5369 h1:Yp0zFEufLz0H7jzffb4UPXijavlyqlYeOg7dcyVUNnQ= -github.com/nareix/joy4 v0.0.0-20200507095837-05a4ffbb5369/go.mod h1:aFJ1ZwLjvHN4yEzE5Bkz8rD8/d8Vlj3UIuvz2yfET7I= +github.com/nareix/joy5 v0.0.0-20200710135721-d57196c8d506 h1:LclgHa9TCVfOXhpmIEyz8wc2ohx0AF/PJu3wLUZZJ8c= +github.com/nareix/joy5 v0.0.0-20200710135721-d57196c8d506/go.mod h1:XmAOs6UJXpNXRwKk+KY/nv5kL6xXYXyellk+A1pTlko= github.com/neelance/astrewrite v0.0.0-20160511093645-99348263ae86/go.mod h1:kHJEU3ofeGjhHklVoIGuVj85JJwZ6kWPaJwCIxgnFmo= github.com/neelance/sourcemap v0.0.0-20151028013722-8c68805598ab/go.mod h1:Qr6/a/Q4r9LP1IltGz7tA7iOK1WonHEYhu1HRBA7ZiM= github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE= @@ -905,9 +905,11 @@ github.com/spaolacci/murmur3 v1.1.0 h1:7c1g84S4BPRrfL5Xrdp6fOJ206sU9y293DDHaoy0b github.com/spaolacci/murmur3 v1.1.0/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA= github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ= github.com/spf13/cast v1.3.0/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE= +github.com/spf13/cobra v0.0.4-0.20190109003409-7547e83b2d85/go.mod h1:1l0Ry5zgKvJasoi3XT1TypsSe7PqH0Sj9dhYf7v3XqQ= github.com/spf13/cobra v0.0.5/go.mod h1:3K3wKZymM7VvHMDS9+Akkh4K60UwM26emMESw8tLCHU= github.com/spf13/jwalterweatherman v1.0.0/go.mod h1:cQK4TGJAtQXfYWX+Ddv3mKDzgVb68N+wFjFa4jdeBTo= github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4= +github.com/spf13/pflag v1.0.4-0.20181223182923-24fa6976df40/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4= github.com/spf13/viper v1.3.2/go.mod h1:ZiWeW+zYFKm7srdB9IoDzzZXaJaI5eL9QjNiN/DMA2s= github.com/src-d/envconfig v1.0.0/go.mod h1:Q9YQZ7BKITldTBnoxsE5gOeB5y66RyPXeue/R4aaNBc= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= diff --git a/utils/utils.go b/utils/utils.go index 0568d97ad..43b211d6e 100644 --- a/utils/utils.go +++ b/utils/utils.go @@ -9,7 +9,7 @@ import ( //GetTemporaryPipePath gets the temporary path for the streampipe.flv file func GetTemporaryPipePath() string { - return filepath.Join(os.TempDir(), "streampipe.ts") + return filepath.Join(os.TempDir(), "streampipe.flv") } //DoesFileExists checks if the file exists