Add GoSRT & improvements (repo-merge)

Commits (Ingo Oppermann):
- Add experimental SRT connection stats and logs
- Hide /config/reload endpoint in reade-only mode
- Add SRT server
- Create v16 in go.mod
- Fix data races, tests, lint, and update dependencies
- Add trailing slash for routed directories (datarhei/restreamer#340)
- Allow relative URLs in content in static routes

Co-Authored-By: Ingo Oppermann <[email protected]>

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Core
 
+#### Core v16.8.0 > ?
+
+-   Add experimental SRT connection stats and logs
+-   Hide /config/reload endpoint in reade-only mode
+-   Add SRT server (datarhei/gosrt)
+-   Create v16 in go.mod
+-   Fix data races, tests, lint, and update dependencies
+-   Add trailing slash for routed directories (datarhei/restreamer#340)
+-   Allow relative URLs in content in static routes
+
 #### Core v16.7.2 > v16.8.0
 
 -   Add purge_on_delete function

app/api/api.go

@@ -15,25 +15,26 @@ import (
 	"sync"
 	"time"
 
-	"github.com/datarhei/core/app"
-	"github.com/datarhei/core/config"
-	"github.com/datarhei/core/ffmpeg"
-	"github.com/datarhei/core/http"
-	"github.com/datarhei/core/http/cache"
-	"github.com/datarhei/core/http/jwt"
-	"github.com/datarhei/core/http/router"
-	"github.com/datarhei/core/io/fs"
-	"github.com/datarhei/core/log"
-	"github.com/datarhei/core/math/rand"
-	"github.com/datarhei/core/monitor"
-	"github.com/datarhei/core/net"
-	"github.com/datarhei/core/prometheus"
-	"github.com/datarhei/core/restream"
-	"github.com/datarhei/core/restream/store"
-	"github.com/datarhei/core/rtmp"
-	"github.com/datarhei/core/service"
-	"github.com/datarhei/core/session"
-	"github.com/datarhei/core/update"
+	"github.com/datarhei/core/v16/app"
+	"github.com/datarhei/core/v16/config"
+	"github.com/datarhei/core/v16/ffmpeg"
+	"github.com/datarhei/core/v16/http"
+	"github.com/datarhei/core/v16/http/cache"
+	"github.com/datarhei/core/v16/http/jwt"
+	"github.com/datarhei/core/v16/http/router"
+	"github.com/datarhei/core/v16/io/fs"
+	"github.com/datarhei/core/v16/log"
+	"github.com/datarhei/core/v16/math/rand"
+	"github.com/datarhei/core/v16/monitor"
+	"github.com/datarhei/core/v16/net"
+	"github.com/datarhei/core/v16/prometheus"
+	"github.com/datarhei/core/v16/restream"
+	"github.com/datarhei/core/v16/restream/store"
+	"github.com/datarhei/core/v16/rtmp"
+	"github.com/datarhei/core/v16/service"
+	"github.com/datarhei/core/v16/session"
+	"github.com/datarhei/core/v16/srt"
+	"github.com/datarhei/core/v16/update"
 
 	"golang.org/x/crypto/acme/autocert"
 )
@@ -64,6 +65,7 @@ type api struct {
 	diskfs        fs.Filesystem
 	memfs         fs.Filesystem
 	rtmpserver    rtmp.Server
+	srtserver     srt.Server
 	metrics       monitor.HistoryMonitor
 	prom          prometheus.Metrics
 	service       service.Service
@@ -76,9 +78,6 @@ type api struct {
 
 	errorChan chan error
 
-	startOnce sync.Once
-	stopOnce  sync.Once
-
 	gcTickerStop context.CancelFunc
 
 	log struct {
@@ -89,6 +88,7 @@ type api struct {
 			main    log.Logger
 			sidecar log.Logger
 			rtmp    log.Logger
+			srt     log.Logger
 		}
 	}
 
@@ -319,6 +319,11 @@ func (a *api) start() error {
 			return fmt.Errorf("unable to register session collector: %w", err)
 		}
 
+		srt, err := sessions.Register("srt", config)
+		if err != nil {
+			return fmt.Errorf("unable to register session collector: %w", err)
+		}
+
 		if _, err := sessions.Register("http", config); err != nil {
 			return fmt.Errorf("unable to register session collector: %w", err)
 		}
@@ -333,13 +338,20 @@ func (a *api) start() error {
 		}
 
 		hls.AddCompanion(rtmp)
+		hls.AddCompanion(srt)
 		hls.AddCompanion(ffmpeg)
 
 		rtmp.AddCompanion(hls)
 		rtmp.AddCompanion(ffmpeg)
+		rtmp.AddCompanion(srt)
+
+		srt.AddCompanion(hls)
+		srt.AddCompanion(ffmpeg)
+		srt.AddCompanion(rtmp)
 
 		ffmpeg.AddCompanion(hls)
 		ffmpeg.AddCompanion(rtmp)
+		ffmpeg.AddCompanion(srt)
 	} else {
 		sessions, _ := session.New(session.Config{})
 		a.sessions = sessions
@@ -674,7 +686,6 @@ func (a *api) start() error {
 		}
 
 		rtmpserver, err := rtmp.New(config)
-
 		if err != nil {
 			return fmt.Errorf("unable to create RMTP server: %w", err)
 		}
@@ -683,6 +694,25 @@ func (a *api) start() error {
 		a.rtmpserver = rtmpserver
 	}
 
+	if cfg.SRT.Enable {
+		config := srt.Config{
+			Addr:         cfg.SRT.Address,
+			Passphrase:   cfg.SRT.Passphrase,
+			Token:        cfg.SRT.Token,
+			Logger:       a.log.logger.core.WithComponent("SRT").WithField("address", cfg.SRT.Address),
+			Collector:    a.sessions.Collector("srt"),
+			SRTLogTopics: []string{"handshake", "connection"},
+		}
+
+		srtserver, err := srt.New(config)
+		if err != nil {
+			return fmt.Errorf("unable to create SRT server: %w", err)
+		}
+
+		a.log.logger.srt = config.Logger
+		a.srtserver = srtserver
+	}
+
 	logcontext := "HTTP"
 	if cfg.TLS.Enable {
 		logcontext = "HTTPS"
@@ -733,6 +763,7 @@ func (a *api) start() error {
 			Origins: cfg.Storage.CORS.Origins,
 		},
 		RTMP:     a.rtmpserver,
+		SRT:      a.srtserver,
 		JWT:      a.httpjwt,
 		Config:   a.config.store,
 		Cache:    a.cache,
@@ -794,6 +825,7 @@ func (a *api) start() error {
 				Origins: cfg.Storage.CORS.Origins,
 			},
 			RTMP:     a.rtmpserver,
+			SRT:      a.srtserver,
 			JWT:      a.httpjwt,
 			Config:   a.config.store,
 			Cache:    a.cache,
@@ -888,6 +920,34 @@ func (a *api) start() error {
 		}()
 	}
 
+	if a.srtserver != nil {
+		wgStart.Add(1)
+		a.wgStop.Add(1)
+
+		go func() {
+			logger := a.log.logger.srt
+
+			defer func() {
+				logger.Info().Log("Server exited")
+				a.wgStop.Done()
+			}()
+
+			wgStart.Done()
+
+			var err error
+
+			logger.Info().Log("Server started")
+			err = a.srtserver.ListenAndServe()
+			if err != nil && err != srt.ErrServerClosed {
+				err = fmt.Errorf("SRT server: %w", err)
+			} else {
+				err = nil
+			}
+
+			sendError(err)
+		}()
+	}
+
 	wgStart.Add(1)
 	a.wgStop.Add(1)
 
@@ -1031,6 +1091,14 @@ func (a *api) stop() {
 		a.cache = nil
 	}
 
+	// Stop the SRT server
+	if a.srtserver != nil {
+		a.log.logger.srt.Info().Log("Stopping ...")
+
+		a.srtserver.Close()
+		a.srtserver = nil
+	}
+
 	// Stop the RTMP server
 	if a.rtmpserver != nil {
 		a.log.logger.rtmp.Info().Log("Stopping ...")

app/import/import.go

@@ -15,12 +15,12 @@ import (
 	"strings"
 	"time"
 
-	"github.com/datarhei/core/encoding/json"
-	"github.com/datarhei/core/ffmpeg"
-	"github.com/datarhei/core/ffmpeg/skills"
-	"github.com/datarhei/core/restream"
-	"github.com/datarhei/core/restream/app"
-	"github.com/datarhei/core/restream/store"
+	"github.com/datarhei/core/v16/encoding/json"
+	"github.com/datarhei/core/v16/ffmpeg"
+	"github.com/datarhei/core/v16/ffmpeg/skills"
+	"github.com/datarhei/core/v16/restream"
+	"github.com/datarhei/core/v16/restream/app"
+	"github.com/datarhei/core/v16/restream/store"
 
 	"github.com/google/uuid"
 )

app/import/import_test.go

@@ -6,8 +6,8 @@ import (
 	"os"
 	"testing"
 
-	"github.com/datarhei/core/encoding/json"
-	"github.com/datarhei/core/restream/store"
+	"github.com/datarhei/core/v16/encoding/json"
+	"github.com/datarhei/core/v16/restream/store"
 
 	"github.com/stretchr/testify/require"
 )

app/import/main.go

@@ -3,9 +3,9 @@ package main
 import (
 	"os"
 
-	"github.com/datarhei/core/config"
-	"github.com/datarhei/core/log"
-	"github.com/datarhei/core/restream/store"
+	"github.com/datarhei/core/v16/config"
+	"github.com/datarhei/core/v16/log"
+	"github.com/datarhei/core/v16/restream/store"
 
 	_ "github.com/joho/godotenv/autoload"
 )

config/config.go

@@ -8,7 +8,7 @@ import (
 	"os"
 	"time"
 
-	"github.com/datarhei/core/math/rand"
+	"github.com/datarhei/core/v16/math/rand"
 
 	haikunator "github.com/atrox/haikunatorgo/v2"
 	"github.com/google/uuid"
@@ -135,6 +135,12 @@ type Data struct {
 		App       string `json:"app"`
 		Token     string `json:"token"`
 	} `json:"rtmp"`
+	SRT struct {
+		Enable     bool   `json:"enable"`
+		Address    string `json:"address"`
+		Passphrase string `json:"passphrase"`
+		Token      string `json:"token"`
+	} `json:"srt"`
 	FFmpeg struct {
 		Binary       string `json:"binary"`
 		MaxProcesses int64  `json:"max_processes"`
@@ -227,6 +233,7 @@ func NewConfigFrom(d *Config) *Config {
 	data.TLS = d.TLS
 	data.Storage = d.Storage
 	data.RTMP = d.RTMP
+	data.SRT = d.SRT
 	data.FFmpeg = d.FFmpeg
 	data.Playout = d.Playout
 	data.Debug = d.Debug
@@ -339,6 +346,12 @@ func (d *Config) init() {
 	d.val(newStringValue(&d.RTMP.App, "/"), "rtmp.app", "CORE_RTMP_APP", nil, "RTMP app for publishing", false, false)
 	d.val(newStringValue(&d.RTMP.Token, ""), "rtmp.token", "CORE_RTMP_TOKEN", nil, "RTMP token for publishing and playing", false, true)
 
+	// SRT
+	d.val(newBoolValue(&d.SRT.Enable, false), "srt.enable", "CORE_SRT_ENABLE", nil, "Enable SRT server", false, false)
+	d.val(newAddressValue(&d.SRT.Address, ":6000"), "srt.address", "CORE_SRT_ADDRESS", nil, "SRT server listen address", false, false)
+	d.val(newStringValue(&d.SRT.Passphrase, ""), "srt.passphrase", "CORE_SRT_PASSPHRASE", nil, "SRT encryption passphrase", false, true)
+	d.val(newStringValue(&d.SRT.Token, ""), "srt.token", "CORE_SRT_TOKEN", nil, "SRT token for publishing and playing", false, true)
+
 	// FFmpeg
 	d.val(newExecValue(&d.FFmpeg.Binary, "ffmpeg"), "ffmpeg.binary", "CORE_FFMPEG_BINARY", nil, "Path to ffmpeg binary", true, false)
 	d.val(newInt64Value(&d.FFmpeg.MaxProcesses, 0), "ffmpeg.max_processes", "CORE_FFMPEG_MAXPROCESSES", nil, "Max. allowed simultaneously running ffmpeg instances, 0 for unlimited", false, false)

config/json.go

@@ -8,8 +8,8 @@ import (
 	"path/filepath"
 	"time"
 
-	"github.com/datarhei/core/encoding/json"
-	"github.com/datarhei/core/io/file"
+	"github.com/datarhei/core/v16/encoding/json"
+	"github.com/datarhei/core/v16/io/file"
 )
 
 type jsonStore struct {

config/types.go

@@ -13,7 +13,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/datarhei/core/http/cors"
+	"github.com/datarhei/core/v16/http/cors"
 )
 
 type value interface {