Re-work RTMP for #34

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/joy5/av"
-	"github.com/nareix/joy4/format/ts"
+	"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/joy5/codec/h264"
-	"github.com/nareix/joy4/format"
+	"github.com/nareix/joy5/format/rtmp"
-	"github.com/nareix/joy4/format/rtmp"
 )
 
 var (
@@ -26,36 +27,53 @@ var (
 	_isConnected = false
 )
 
-func init() {
+var _transcoder ffmpeg.Transcoder
-	format.RegisterAll()
+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
+	w := flv.NewMuxer(f)
-	// into the stream so frequently?
+
-	ticker := time.NewTicker(500 * time.Millisecond)
+	for {
-	go func() {
+		pkt, err := c.ReadPacket()
-		for {
+		if err != nil {
-			select {
+			if err == io.EOF {
-			case <-ticker.C:
+				handleDisconnect(nc)
-				error := connCheck(conn.NetConn())
-				if error == io.EOF {
-					handleDisconnect(conn)
-				}
 			}
 		}
-	}()
-	muxer := ts.NewMuxer(f)
-	avutil.CopyFile(muxer, conn)
-}
 
-// Proactively check if the RTMP connection is still active or not.
+		if pkt.Type == av.H264 {
-// Taken from https://stackoverflow.com/a/58664631.
+			nalus, _ := h264.SplitNALUs(pkt.Data)
-func connCheck(conn net.Conn) error {
+			annexb := h264.JoinNALUsAnnexb(nalus)
-	var sysErr error = nil
+			avcc := h264.JoinNALUsAVCC([][]byte{annexb})
-	rc, err := conn.(syscall.Conn).SyscallConn()
+			pkt.Data = avcc
-	if err != nil {
+		}
-		return err
+
-	}
+		if err := w.WritePacket(pkt); err != nil {
-	err = rc.Read(func(fd uintptr) bool {
+			panic(err)
-		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
 		}
-		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()
 }
 

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

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/joy5 v0.0.0-20200710135721-d57196c8d506 h1:LclgHa9TCVfOXhpmIEyz8wc2ohx0AF/PJu3wLUZZJ8c=
-github.com/nareix/joy4 v0.0.0-20200507095837-05a4ffbb5369/go.mod h1:aFJ1ZwLjvHN4yEzE5Bkz8rD8/d8Vlj3UIuvz2yfET7I=
+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=

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