feat: SRTLA receiver support

docs/2-features/25-srt-specific-features.md

@@ -16,3 +16,23 @@ Where:
 - key `r` contains the path
 - key `u` contains the username
 - key `s` contains the password
+
+## SRTLA (SRT Link Aggregation)
+
+SRTLA is an extension of SRT that bonds multiple network connections (e.g. cellular + Wi-Fi) into a single reliable stream. This is commonly used by mobile streaming devices like BELABOX, IRL Pro, and Moblin.
+
+When enabled, the server listens for SRTLA connections on a separate UDP port and transparently proxies them to the local SRT server.
+
+To enable SRTLA, set the following in the configuration file:
+
+```yaml
+srtla: true
+srtlaAddress: :8891
+```
+
+SRTLA clients connect to port `8891` and register multiple network links. The server aggregates incoming data packets and forwards them to the SRT server on port `8890`. SRT ACK packets are broadcast to all registered links for timely delivery, while other responses are sent only to the most recently active link.
+
+Configuration reference:
+
+- `srtla` — enable or disable the SRTLA listener (default: `true`)
+- `srtlaAddress` — address of the SRTLA listener (default: `:8891`)

docs/3-publish/02-srt-clients.md

@@ -19,3 +19,17 @@ If you need to use the standard stream ID syntax instead of the custom one in us
 If you want to publish a stream by using a client in listening mode (i.e. with `mode=listener` appended to the URL), read the next section.
 
 Some clients that can publish with SRT are [FFmpeg](17-ffmpeg.md), [GStreamer](18-gstreamer.md), [OBS Studio](19-obs-studio.md).
+
+## Publishing with SRTLA
+
+SRTLA (SRT Link Aggregation) allows bonding multiple network connections for improved reliability during mobile streaming. SRTLA-capable clients (BELABOX, IRL Pro, Moblin) can connect to the SRTLA port (default `:8891`) and register multiple links that are aggregated into a single SRT stream.
+
+To publish via SRTLA, point the sender to the SRTLA port instead of the SRT port:
+
+```
+srtla://yourserver:8891?streamid=publish:mystream
+```
+
+The SRTLA receiver bonds all registered connections and forwards the aggregated stream to the local SRT server. No additional SRT configuration is needed — path selection, authentication, and codec handling work identically to direct SRT publishing.
+
+See [SRTLA configuration](../2-features/25-srt-specific-features.md#srtla-srt-link-aggregation) for server-side setup.

internal/conf/conf.go

@@ -407,6 +407,10 @@ type Conf struct {
 	MoQHTTPS2Address  *string    `json:"moqHTTPS2Address,omitempty" deprecated:"true"`
 	MoQHTTPS3Address  *string    `json:"moqHTTPS3Address,omitempty" deprecated:"true"`
 
+	// SRTLA server
+	SRTLA        bool   `json:"srtla"`
+	SRTLAAddress string `json:"srtlaAddress"`
+
 	// Record (deprecated)
 	Record                *bool         `json:"record,omitempty" deprecated:"true"`
 	RecordPath            *string       `json:"recordPath,omitempty" deprecated:"true"`
@@ -543,6 +547,10 @@ func (conf *Conf) setDefaults() {
 	conf.MoQServerCert = "auto.crt"
 	conf.MoQAllowOrigins = []string{"*"}
 
+	// SRTLA server
+	conf.SRTLA = true
+	conf.SRTLAAddress = ":8891"
+
 	conf.PathDefaults.setDefaults()
 }
 
@@ -1058,6 +1066,12 @@ func (conf *Conf) Validate(l logger.Writer) error {
 
 	// Record (deprecated)
 
+	if conf.SRTLA {
+		if conf.SRTLAAddress == "" {
+			return fmt.Errorf("'srtlaAddress' must be set when SRTLA is enabled")
+		}
+	}
+
 	if conf.Record != nil {
 		l.Log(logger.Warn, "parameter 'record' is deprecated "+
 			"and has been replaced with 'pathDefaults.record'")

internal/core/core.go

@@ -35,6 +35,7 @@ import (
 	"github.com/bluenviron/mediamtx/internal/servers/rtmp"
 	"github.com/bluenviron/mediamtx/internal/servers/rtsp"
 	"github.com/bluenviron/mediamtx/internal/servers/srt"
+	"github.com/bluenviron/mediamtx/internal/servers/srtla"
 	"github.com/bluenviron/mediamtx/internal/servers/webrtc"
 	"github.com/bluenviron/mediamtx/internal/upgrade"
 )
@@ -127,6 +128,7 @@ type Core struct {
 	webRTCServer    *webrtc.Server
 	srtServer       *srt.Server
 	moqServer       *moq.Server
+	srtlaServer     *srtla.Server
 	api             *api.API
 	confWatcher     *confwatcher.ConfWatcher
 
@@ -723,6 +725,23 @@ func (p *Core) createResources(initial bool) error {
 		p.moqServer = i
 	}
 
+	if p.conf.SRTLA &&
+		p.conf.SRT &&
+		p.srtlaServer == nil {
+		i := &srtla.Server{
+			Address:    p.conf.SRTLAAddress,
+			SRTAddress: p.conf.SRTAddress,
+			Metrics:    p.metrics,
+			Parent:     p,
+		}
+		err = i.Initialize()
+		if err != nil {
+			return err
+		}
+		p.srtlaServer = i
+		p.srtServer.SRTLALinker = i
+	}
+
 	if p.conf.API &&
 		p.api == nil {
 		i := &api.API{
@@ -1023,6 +1042,14 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) {
 		closePathManager ||
 		closeLogger
 
+	closeSRTLAServer := newConf == nil ||
+		newConf.SRTLA != p.conf.SRTLA ||
+		newConf.SRTLAAddress != p.conf.SRTLAAddress ||
+		newConf.SRTAddress != p.conf.SRTAddress ||
+		newConf.SRT != p.conf.SRT ||
+		closeSRTServer ||
+		closeLogger
+
 	closeAPI := newConf == nil ||
 		newConf.API != p.conf.API ||
 		newConf.APIAddress != p.conf.APIAddress ||
@@ -1060,6 +1087,11 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) {
 		}
 	}
 
+	if closeSRTLAServer && p.srtlaServer != nil {
+		p.srtlaServer.Close()
+		p.srtlaServer = nil
+	}
+
 	if closeSRTServer && p.srtServer != nil {
 		p.srtServer.Close()
 		p.srtServer = nil

internal/core/metrics_test.go

@@ -241,6 +241,12 @@ srt_conns_packets_send_loss_rate 0
 srt_conns_packets_received_loss_rate 0
 srt_conns_outbound_frames_discarded 0
 
+# SRTLA groups
+srtla_groups 0
+srtla_groups_conns_active 0
+srtla_groups_bytes_received 0
+srtla_groups_bytes_forwarded 0
+
 # WebRTC sessions
 webrtc_sessions 0
 webrtc_sessions_inbound_bytes 0

internal/defs/api_srtla.go

@@ -0,0 +1,31 @@
+package defs
+
+// SRTLAGroupInfo contains information about an SRTLA group for correlation purposes.
+type SRTLAGroupInfo struct {
+	Path           string
+	ConnsActive    int
+	BytesReceived  uint64
+	BytesForwarded uint64
+}
+
+// SRTLALinker allows the SRT server to correlate connections with SRTLA groups.
+type SRTLALinker interface {
+	// SetGroupPath sets the stream path on the SRTLA group identified by the SRT connection's remote address.
+	SetGroupPath(srtConnAddr string, path string)
+	// CloseGroupByAddr closes the SRTLA group associated with the given SRT connection address.
+	CloseGroupByAddr(srtConnAddr string)
+}
+
+// APISRTLAServer contains methods used by the Metrics server.
+type APISRTLAServer interface {
+	APISRTLAGroupsList() []APISRTLAGroup
+}
+
+// APISRTLAGroup is an SRTLA group entry for metrics/API.
+type APISRTLAGroup struct {
+	ID             string `json:"id"`
+	Path           string `json:"path"`
+	ConnsActive    int    `json:"connsActive"`
+	BytesReceived  uint64 `json:"bytesReceived"`
+	BytesForwarded uint64 `json:"bytesForwarded"`
+}

internal/metrics/metrics.go

@@ -83,6 +83,7 @@ const (
 	metricsTypeRTMPConns      metricsType = "rtmp_conns"
 	metricsTypeRTMPSConns     metricsType = "rtmps_conns"
 	metricsTypeSRTConns       metricsType = "srt_conns"
+	metricsTypeSRTLAGroups    metricsType = "srtla_groups"
 	metricsTypeWebRTCSessions metricsType = "webrtc_sessions"
 	metricsTypeMoQSessions    metricsType = "moq_sessions"
 )
@@ -118,6 +119,7 @@ type Metrics struct {
 	rtmpServer   defs.APIRTMPServer
 	rtmpsServer  defs.APIRTMPServer
 	srtServer    defs.APISRTServer
+	srtlaServer  defs.APISRTLAServer
 	webRTCServer defs.APIWebRTCServer
 	moqServer    defs.APIMoQServer
 }
@@ -222,6 +224,7 @@ func (m *Metrics) onMetrics(ctx *gin.Context) {
 	rtmpServer := m.rtmpServer
 	rtmpsServer := m.rtmpsServer
 	srtServer := m.srtServer
+	srtlaServer := m.srtlaServer
 	webRTCServer := m.webRTCServer
 	moqServer := m.moqServer
 	m.mutex.RUnlock()
@@ -918,6 +921,34 @@ func (m *Metrics) onMetrics(ctx *gin.Context) {
 		}
 	}
 
+	if !interfaceIsEmpty(srtlaServer) &&
+		(typ == "" || typ == metricsTypeSRTLAGroups) &&
+		!anyFilterActive {
+		data := srtlaServer.APISRTLAGroupsList()
+		if len(data) != 0 {
+			out.WriteString("# SRTLA groups\n")
+			for _, i := range data {
+				ta := tags(map[string]string{
+					"id":   i.ID,
+					"path": i.Path,
+				})
+
+				metric(&out, "srtla_groups", ta, 1)
+				metric(&out, "srtla_groups_conns_active", ta, int64(i.ConnsActive))
+				metric(&out, "srtla_groups_bytes_received", ta, int64(i.BytesReceived))
+				metric(&out, "srtla_groups_bytes_forwarded", ta, int64(i.BytesForwarded))
+			}
+			out.WriteString("\n")
+		} else {
+			out.WriteString("# SRTLA groups\n")
+			metric(&out, "srtla_groups", "", 0)
+			metric(&out, "srtla_groups_conns_active", "", 0)
+			metric(&out, "srtla_groups_bytes_received", "", 0)
+			metric(&out, "srtla_groups_bytes_forwarded", "", 0)
+			out.WriteString("\n")
+		}
+	}
+
 	if !interfaceIsEmpty(webRTCServer) &&
 		(typ == "" || typ == metricsTypeWebRTCSessions) &&
 		(!anyFilterActive || webrtcSessionFilter != "") {
@@ -1080,6 +1111,13 @@ func (m *Metrics) SetSRTServer(s defs.APISRTServer) {
 	m.srtServer = s
 }
 
+// SetSRTLAServer is called by core.
+func (m *Metrics) SetSRTLAServer(s defs.APISRTLAServer) {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+	m.srtlaServer = s
+}
+
 // SetWebRTCServer is called by core.
 func (m *Metrics) SetWebRTCServer(s defs.APIWebRTCServer) {
 	m.mutex.Lock()

internal/servers/srt/conn.go

@@ -113,6 +113,10 @@ func (c *conn) run() { //nolint:dupl
 
 	c.ctxCancel()
 
+	if c.parent.SRTLALinker != nil {
+		c.parent.SRTLALinker.CloseGroupByAddr(c.connReq.RemoteAddr().String())
+	}
+
 	c.parent.closeConn(c)
 
 	c.Log(logger.Info, "closed: %v", err)
@@ -248,6 +252,10 @@ func (c *conn) runPublishReader(sconn srt.Conn, streamID *streamID, pathConf *co
 	c.sconn = sconn
 	c.mutex.Unlock()
 
+	if c.parent.SRTLALinker != nil {
+		c.parent.SRTLALinker.SetGroupPath(c.connReq.RemoteAddr().String(), streamID.path)
+	}
+
 	for {
 		err = r.Read()
 		if err != nil {
@@ -311,6 +319,10 @@ func (c *conn) runRead(streamID *streamID) error {
 	c.sconn = sconn
 	c.mutex.Unlock()
 
+	if c.parent.SRTLALinker != nil {
+		c.parent.SRTLALinker.SetGroupPath(c.connReq.RemoteAddr().String(), streamID.path)
+	}
+
 	c.Log(logger.Info, "is reading from path '%s', %s",
 		res.Path.Name(), defs.FormatsInfo(r.Formats()))
 

internal/servers/srt/server.go

@@ -83,6 +83,7 @@ type Server struct {
 	RunOnConnectRestart bool
 	RunOnDisconnect     string
 	ExternalCmdPool     *externalcmd.Pool
+	SRTLALinker         defs.SRTLALinker
 	Metrics             serverMetrics
 	PathManager         serverPathManager
 	Parent              serverParent