Cleanup + poll connection for disconnected state. For #34
This commit is contained in:
Gabe Kangas
@@ -1,188 +0,0 @@
|
||||
package rtmp
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"errors"
|
||||
"io"
|
||||
"os"
|
||||
"syscall"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
"github.com/yutopp/go-flv"
|
||||
flvtag "github.com/yutopp/go-flv/tag"
|
||||
yutmp "github.com/yutopp/go-rtmp"
|
||||
rtmpmsg "github.com/yutopp/go-rtmp/message"
|
||||
|
||||
"github.com/gabek/owncast/config"
|
||||
"github.com/gabek/owncast/core"
|
||||
"github.com/gabek/owncast/core/ffmpeg"
|
||||
"github.com/gabek/owncast/utils"
|
||||
)
|
||||
|
||||
var _ yutmp.Handler = (*Handler)(nil)
|
||||
|
||||
// Handler An RTMP connection handler
|
||||
type Handler struct {
|
||||
yutmp.DefaultHandler
|
||||
flvFile *os.File
|
||||
flvEnc *flv.Encoder
|
||||
}
|
||||
|
||||
//OnServe handles the "OnServe" of the rtmp service
|
||||
func (h *Handler) OnServe(conn *yutmp.Conn) {
|
||||
}
|
||||
|
||||
//OnConnect handles the "OnConnect" of the rtmp service
|
||||
func (h *Handler) OnConnect(timestamp uint32, cmd *rtmpmsg.NetConnectionConnect) error {
|
||||
// log.Printf("OnConnect: %#v", cmd)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
//OnCreateStream handles the "OnCreateStream" of the rtmp service
|
||||
func (h *Handler) OnCreateStream(timestamp uint32, cmd *rtmpmsg.NetConnectionCreateStream) error {
|
||||
// log.Printf("OnCreateStream: %#v", cmd)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
//OnPublish handles the "OnPublish" of the rtmp service
|
||||
func (h *Handler) OnPublish(timestamp uint32, cmd *rtmpmsg.NetStreamPublish) error {
|
||||
// log.Printf("OnPublish: %#v", cmd)
|
||||
log.Trace("Incoming stream connected.")
|
||||
|
||||
if cmd.PublishingName != config.Config.VideoSettings.StreamingKey {
|
||||
return errors.New("invalid streaming key; rejecting incoming stream")
|
||||
}
|
||||
|
||||
if _isConnected {
|
||||
return errors.New("stream already running; can not overtake an existing stream")
|
||||
}
|
||||
|
||||
// Record streams as FLV
|
||||
p := utils.GetTemporaryPipePath()
|
||||
syscall.Mkfifo(p, 0666)
|
||||
|
||||
f, err := os.OpenFile(p, os.O_RDWR, os.ModeNamedPipe)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
h.flvFile = f
|
||||
|
||||
enc, err := flv.NewEncoder(f, flv.FlagsAudio|flv.FlagsVideo)
|
||||
if err != nil {
|
||||
_ = f.Close()
|
||||
return err
|
||||
}
|
||||
h.flvEnc = enc
|
||||
|
||||
transcoder := ffmpeg.NewTranscoder()
|
||||
go transcoder.Start()
|
||||
|
||||
_isConnected = true
|
||||
core.SetStreamAsConnected()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
//OnSetDataFrame handles the setting of the data frame
|
||||
func (h *Handler) OnSetDataFrame(timestamp uint32, data *rtmpmsg.NetStreamSetDataFrame) error {
|
||||
r := bytes.NewReader(data.Payload)
|
||||
|
||||
var script flvtag.ScriptData
|
||||
if err := flvtag.DecodeScriptData(r, &script); err != nil {
|
||||
log.Printf("Failed to decode script data: Err = %+v", err)
|
||||
return nil // ignore
|
||||
}
|
||||
|
||||
// log.Printf("SetDataFrame: Script = %#v", script)
|
||||
|
||||
if err := h.flvEnc.Encode(&flvtag.FlvTag{
|
||||
TagType: flvtag.TagTypeScriptData,
|
||||
Timestamp: timestamp,
|
||||
Data: &script,
|
||||
}); err != nil {
|
||||
log.Printf("Failed to write script data: Err = %+v", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
//OnAudio handles when we get audio from the rtmp service
|
||||
func (h *Handler) OnAudio(timestamp uint32, payload io.Reader) error {
|
||||
var audio flvtag.AudioData
|
||||
if err := flvtag.DecodeAudioData(payload, &audio); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
flvBody := new(bytes.Buffer)
|
||||
if _, err := io.Copy(flvBody, audio.Data); err != nil {
|
||||
return err
|
||||
}
|
||||
audio.Data = flvBody
|
||||
|
||||
// log.Printf("FLV Audio Data: Timestamp = %d, SoundFormat = %+v, SoundRate = %+v, SoundSize = %+v, SoundType = %+v, AACPacketType = %+v, Data length = %+v",
|
||||
// timestamp,
|
||||
// audio.SoundFormat,
|
||||
// audio.SoundRate,
|
||||
// audio.SoundSize,
|
||||
// audio.SoundType,
|
||||
// audio.AACPacketType,
|
||||
// len(flvBody.Bytes()),
|
||||
// )
|
||||
|
||||
if err := h.flvEnc.Encode(&flvtag.FlvTag{
|
||||
TagType: flvtag.TagTypeAudio,
|
||||
Timestamp: timestamp,
|
||||
Data: &audio,
|
||||
}); err != nil {
|
||||
log.Printf("Failed to write audio: Err = %+v", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
//OnVideo handles when we video from the rtmp service
|
||||
func (h *Handler) OnVideo(timestamp uint32, payload io.Reader) error {
|
||||
var video flvtag.VideoData
|
||||
if err := flvtag.DecodeVideoData(payload, &video); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
flvBody := new(bytes.Buffer)
|
||||
if _, err := io.Copy(flvBody, video.Data); err != nil {
|
||||
return err
|
||||
}
|
||||
video.Data = flvBody
|
||||
|
||||
// log.Printf("FLV Video Data: Timestamp = %d, FrameType = %+v, CodecID = %+v, AVCPacketType = %+v, CT = %+v, Data length = %+v",
|
||||
// timestamp,
|
||||
// video.FrameType,
|
||||
// video.CodecID,
|
||||
// video.AVCPacketType,
|
||||
// video.CompositionTime,
|
||||
// len(flvBody.Bytes()),
|
||||
// )
|
||||
|
||||
if err := h.flvEnc.Encode(&flvtag.FlvTag{
|
||||
TagType: flvtag.TagTypeVideo,
|
||||
Timestamp: timestamp,
|
||||
Data: &video,
|
||||
}); err != nil {
|
||||
log.Printf("Failed to write video: Err = %+v", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
//OnClose handles the closing of the rtmp connection
|
||||
func (h *Handler) OnClose() {
|
||||
log.Printf("OnClose of the rtmp service")
|
||||
|
||||
if h.flvFile != nil {
|
||||
_ = h.flvFile.Close()
|
||||
}
|
||||
|
||||
_isConnected = false
|
||||
core.SetStreamAsDisconnected()
|
||||
}
|
||||
@@ -1,12 +1,15 @@
|
||||
package rtmp
|
||||
|
||||
import (
|
||||
"io"
|
||||
"net"
|
||||
"os"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/Seize/joy4/av/avutil"
|
||||
"github.com/Seize/joy4/format/ts"
|
||||
"github.com/nareix/joy4/av/avutil"
|
||||
"github.com/nareix/joy4/format/ts"
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/gabek/owncast/config"
|
||||
@@ -14,8 +17,8 @@ import (
|
||||
"github.com/gabek/owncast/core/ffmpeg"
|
||||
"github.com/gabek/owncast/utils"
|
||||
|
||||
"github.com/Seize/joy4/format"
|
||||
"github.com/Seize/joy4/format/rtmp"
|
||||
"github.com/nareix/joy4/format"
|
||||
"github.com/nareix/joy4/format/rtmp"
|
||||
)
|
||||
|
||||
var (
|
||||
@@ -29,7 +32,6 @@ func init() {
|
||||
|
||||
//Start starts the rtmp service, listening on port 1935
|
||||
func Start() {
|
||||
|
||||
port := 1935
|
||||
server := &rtmp.Server{}
|
||||
|
||||
@@ -43,12 +45,11 @@ func Start() {
|
||||
}
|
||||
|
||||
func handlePublish(conn *rtmp.Conn) {
|
||||
// Commented out temporarily because I have no way to set _isConnected to false after RTMP is closed.
|
||||
// if _isConnected {
|
||||
// log.Errorln("stream already running; can not overtake an existing stream")
|
||||
// conn.Close()
|
||||
// return
|
||||
// }
|
||||
if _isConnected {
|
||||
log.Errorln("stream already running; can not overtake an existing stream")
|
||||
conn.Close()
|
||||
return
|
||||
}
|
||||
|
||||
streamingKeyComponents := strings.Split(conn.URL.Path, "/")
|
||||
streamingKey := streamingKeyComponents[len(streamingKeyComponents)-1]
|
||||
@@ -58,6 +59,8 @@ func handlePublish(conn *rtmp.Conn) {
|
||||
return
|
||||
}
|
||||
|
||||
log.Println("Incoming RTMP connected.")
|
||||
|
||||
pipePath := utils.GetTemporaryPipePath()
|
||||
syscall.Mkfifo(pipePath, 0666)
|
||||
transcoder := ffmpeg.NewTranscoder()
|
||||
@@ -71,10 +74,59 @@ func handlePublish(conn *rtmp.Conn) {
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
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
|
||||
}
|
||||
return true
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return sysErr
|
||||
}
|
||||
|
||||
func handleDisconnect(conn *rtmp.Conn) {
|
||||
log.Println("RTMP disconnected.")
|
||||
conn.Close()
|
||||
_isConnected = false
|
||||
core.SetStreamAsDisconnected()
|
||||
}
|
||||
|
||||
//IsConnected gets whether there is an rtmp connection or not
|
||||
//this is only a getter since it is controlled by the rtmp handler
|
||||
func IsConnected() bool {
|
||||
|
||||
Reference in New Issue
Block a user