feat: move transcoder under video

core/core.go

@@ -12,11 +12,11 @@ import (
 	"github.com/owncast/owncast/core/chat"
 	"github.com/owncast/owncast/core/data"
 	"github.com/owncast/owncast/core/rtmp"
-	"github.com/owncast/owncast/core/transcoder"
 	"github.com/owncast/owncast/core/webhooks"
 	"github.com/owncast/owncast/models"
 	"github.com/owncast/owncast/services/notifications"
 	"github.com/owncast/owncast/utils"
+	"github.com/owncast/owncast/video/transcoder"
 	"github.com/owncast/owncast/yp"
 )
 

core/rtmp/broadcaster.go

@@ -1,34 +0,0 @@
-package rtmp
-
-import (
-	"time"
-
-	"github.com/nareix/joy5/format/flv/flvio"
-	"github.com/owncast/owncast/models"
-	log "github.com/sirupsen/logrus"
-)
-
-func setCurrentBroadcasterInfo(t flvio.Tag, remoteAddr string) {
-	data, err := getInboundDetailsFromMetadata(t.DebugFields())
-	if err != nil {
-		log.Traceln("Unable to parse inbound broadcaster details:", err)
-	}
-
-	broadcaster := models.Broadcaster{
-		RemoteAddr: remoteAddr,
-		Time:       time.Now(),
-		StreamDetails: models.InboundStreamDetails{
-			Width:          data.Width,
-			Height:         data.Height,
-			VideoBitrate:   int(data.VideoBitrate),
-			VideoCodec:     getVideoCodec(data.VideoCodec),
-			VideoFramerate: data.VideoFramerate,
-			AudioBitrate:   int(data.AudioBitrate),
-			AudioCodec:     getAudioCodec(data.AudioCodec),
-			Encoder:        data.Encoder,
-			VideoOnly:      data.AudioCodec == nil,
-		},
-	}
-
-	_setBroadcaster(broadcaster)
-}

core/rtmp/rtmp.go

@@ -1,164 +0,0 @@
-package rtmp
-
-import (
-	"fmt"
-	"io"
-	"net"
-	"time"
-
-	"github.com/nareix/joy5/format/flv"
-	"github.com/nareix/joy5/format/flv/flvio"
-	log "github.com/sirupsen/logrus"
-
-	"github.com/nareix/joy5/format/rtmp"
-	"github.com/owncast/owncast/config"
-	"github.com/owncast/owncast/core/data"
-	"github.com/owncast/owncast/models"
-)
-
-var _hasInboundRTMPConnection = false
-
-var (
-	_pipe           *io.PipeWriter
-	_rtmpConnection net.Conn
-)
-
-var (
-	_setStreamAsConnected func(*io.PipeReader)
-	_setBroadcaster       func(models.Broadcaster)
-)
-
-// Start starts the rtmp service, listening on specified RTMP port.
-func Start(setStreamAsConnected func(*io.PipeReader), setBroadcaster func(models.Broadcaster)) {
-	_setStreamAsConnected = setStreamAsConnected
-	_setBroadcaster = setBroadcaster
-
-	port := data.GetRTMPPortNumber()
-	s := rtmp.NewServer()
-	var lis net.Listener
-	var err error
-	if lis, err = net.Listen("tcp", fmt.Sprintf(":%d", port)); err != nil {
-		log.Fatal(err)
-	}
-
-	s.LogEvent = func(c *rtmp.Conn, nc net.Conn, e int) {
-		es := rtmp.EventString[e]
-		log.Traceln("RTMP", nc.LocalAddr(), nc.RemoteAddr(), es)
-	}
-
-	s.HandleConn = HandleConn
-
-	if err != nil {
-		log.Panicln(err)
-	}
-	log.Tracef("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)
-	}
-}
-
-// HandleConn is fired when an inbound RTMP connection takes place.
-func HandleConn(c *rtmp.Conn, nc net.Conn) {
-	c.LogTagEvent = func(isRead bool, t flvio.Tag) {
-		if t.Type == flvio.TAG_AMF0 {
-			log.Tracef("%+v\n", t.DebugFields())
-			setCurrentBroadcasterInfo(t, nc.RemoteAddr().String())
-		}
-	}
-
-	if _hasInboundRTMPConnection {
-		log.Errorln("stream already running; can not overtake an existing stream from", nc.RemoteAddr().String())
-		_ = nc.Close()
-		return
-	}
-
-	accessGranted := false
-	validStreamingKeys := data.GetStreamKeys()
-
-	// If a stream key override was specified then use that instead.
-	if config.TemporaryStreamKey != "" {
-		validStreamingKeys = []models.StreamKey{{Key: config.TemporaryStreamKey}}
-	}
-
-	for _, key := range validStreamingKeys {
-		if secretMatch(key.Key, c.URL.Path) {
-			accessGranted = true
-			break
-		}
-	}
-
-	if !accessGranted {
-		log.Errorln("invalid streaming key; rejecting incoming stream from", nc.RemoteAddr().String())
-		_ = nc.Close()
-		return
-	}
-
-	rtmpOut, rtmpIn := io.Pipe()
-	_pipe = rtmpIn
-	log.Infoln("Inbound stream connected from", nc.RemoteAddr().String())
-	_setStreamAsConnected(rtmpOut)
-
-	_hasInboundRTMPConnection = true
-	_rtmpConnection = nc
-
-	w := flv.NewMuxer(rtmpIn)
-
-	for {
-		if !_hasInboundRTMPConnection {
-			break
-		}
-
-		// If we don't get a readable packet in 10 seconds give up and disconnect
-		if err := _rtmpConnection.SetReadDeadline(time.Now().Add(10 * time.Second)); err != nil {
-			log.Debugln(err)
-		}
-
-		pkt, err := c.ReadPacket()
-
-		// Broadcaster disconnected
-		if err == io.EOF {
-			handleDisconnect(nc)
-			return
-		}
-
-		// Read timeout.  Disconnect.
-		if neterr, ok := err.(net.Error); ok && neterr.Timeout() {
-			log.Debugln("Timeout reading the inbound stream from the broadcaster.  Assuming that they disconnected and ending the stream.")
-			handleDisconnect(nc)
-			return
-		}
-
-		if err := w.WritePacket(pkt); err != nil {
-			log.Errorln("unable to write rtmp packet", err)
-			handleDisconnect(nc)
-			return
-		}
-	}
-}
-
-func handleDisconnect(conn net.Conn) {
-	if !_hasInboundRTMPConnection {
-		return
-	}
-
-	log.Infoln("Inbound stream disconnected.")
-	_ = conn.Close()
-	_ = _pipe.Close()
-	_hasInboundRTMPConnection = false
-}
-
-// Disconnect will force disconnect the current inbound RTMP connection.
-func Disconnect() {
-	if _rtmpConnection == nil {
-		return
-	}
-
-	log.Traceln("Inbound stream disconnect requested.")
-	handleDisconnect(_rtmpConnection)
-}

core/rtmp/utils.go

@@ -1,96 +0,0 @@
-package rtmp
-
-import (
-	"crypto/subtle"
-	"encoding/json"
-	"errors"
-	"fmt"
-	"regexp"
-	"strings"
-
-	"github.com/nareix/joy5/format/flv/flvio"
-	"github.com/owncast/owncast/models"
-	log "github.com/sirupsen/logrus"
-)
-
-const unknownString = "Unknown"
-
-var _getInboundDetailsFromMetadataRE = regexp.MustCompile(`\{(.*?)\}`)
-
-func getInboundDetailsFromMetadata(metadata []interface{}) (models.RTMPStreamMetadata, error) {
-	metadataComponentsString := fmt.Sprintf("%+v", metadata)
-	if !strings.Contains(metadataComponentsString, "onMetaData") {
-		return models.RTMPStreamMetadata{}, errors.New("Not a onMetaData message")
-	}
-
-	submatchall := _getInboundDetailsFromMetadataRE.FindAllString(metadataComponentsString, 1)
-
-	if len(submatchall) == 0 {
-		return models.RTMPStreamMetadata{}, errors.New("unable to parse inbound metadata")
-	}
-
-	metadataJSONString := submatchall[0]
-	var details models.RTMPStreamMetadata
-	err := json.Unmarshal([]byte(metadataJSONString), &details)
-	return details, err
-}
-
-func getAudioCodec(codec interface{}) string {
-	if codec == nil {
-		return "No audio"
-	}
-
-	var codecID float64
-	if assertedCodecID, ok := codec.(float64); ok {
-		codecID = assertedCodecID
-	} else {
-		return codec.(string)
-	}
-
-	switch codecID {
-	case flvio.SOUND_MP3:
-		return "MP3"
-	case flvio.SOUND_AAC:
-		return "AAC"
-	case flvio.SOUND_SPEEX:
-		return "Speex"
-	}
-
-	return unknownString
-}
-
-func getVideoCodec(codec interface{}) string {
-	if codec == nil {
-		return unknownString
-	}
-
-	var codecID float64
-	if assertedCodecID, ok := codec.(float64); ok {
-		codecID = assertedCodecID
-	} else {
-		return codec.(string)
-	}
-
-	switch codecID {
-	case flvio.VIDEO_H264:
-		return "H.264"
-	case flvio.VIDEO_H265:
-		return "H.265"
-	}
-
-	return unknownString
-}
-
-func secretMatch(configStreamKey string, path string) bool {
-	prefix := "/live/"
-
-	if !strings.HasPrefix(path, prefix) {
-		log.Debug("RTMP path does not start with " + prefix)
-		return false // We need the path to begin with $prefix
-	}
-
-	streamingKey := path[len(prefix):] // Remove $prefix
-
-	matches := subtle.ConstantTimeCompare([]byte(streamingKey), []byte(configStreamKey)) == 1
-	return matches
-}

core/rtmp/utils_test.go

@@ -1,35 +0,0 @@
-package rtmp
-
-import "testing"
-
-func Test_secretMatch(t *testing.T) {
-	tests := []struct {
-		name      string
-		streamKey string
-		path      string
-		want      bool
-	}{
-		{"positive", "abc", "/live/abc", true},
-		{"negative", "abc", "/live/def", false},
-		{"positive with numbers", "abc123", "/live/abc123", true},
-		{"negative with numbers", "abc123", "/live/def456", false},
-		{"positive with url chars", "one/two/three", "/live/one/two/three", true},
-		{"negative with url chars", "one/two/three", "/live/four/five/six", false},
-		{"check the entire secret", "three", "/live/one/two/three", false},
-		{"with /live/ in secret", "one/live/three", "/live/one/live/three", true},
-		{"bad path", "anything", "nonsense", false},
-		{"missing secret", "abc", "/live/", false},
-		{"missing secret and missing last slash", "abc", "/live", false},
-		{"streamkey before /live/", "streamkey", "/streamkey/live", false},
-		{"missing /live/", "anything", "/something/else", false},
-		{"stuff before and after /live/", "after", "/before/live/after", false},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if got := secretMatch(tt.streamKey, tt.path); got != tt.want {
-				t.Errorf("secretMatch() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}

core/streamState.go

@@ -12,11 +12,11 @@ import (
 	"github.com/owncast/owncast/core/chat"
 	"github.com/owncast/owncast/core/data"
 	"github.com/owncast/owncast/core/rtmp"
-	"github.com/owncast/owncast/core/transcoder"
 	"github.com/owncast/owncast/core/webhooks"
 	"github.com/owncast/owncast/models"
 	"github.com/owncast/owncast/services/notifications"
 	"github.com/owncast/owncast/utils"
+	"github.com/owncast/owncast/video/transcoder"
 )
 
 // After the stream goes offline this timer fires a full cleanup after N min.

webserver/handlers/adminApiServerConfig.go

@@ -6,9 +6,9 @@ import (
 
 	"github.com/owncast/owncast/config"
 	"github.com/owncast/owncast/core/data"
-	"github.com/owncast/owncast/core/transcoder"
 	"github.com/owncast/owncast/models"
 	"github.com/owncast/owncast/utils"
+	"github.com/owncast/owncast/video/transcoder"
 	"github.com/owncast/owncast/webserver/middleware"
 	log "github.com/sirupsen/logrus"
 )