Implement adaptive streaming support with main and sub stream selection based on viewer quality requests

2026-08-23 15:08:32 +00:00 · 2026-06-27 14:30:20 +00:00
parent 2dd9d50954
commit 675a8a4fb9
9 changed files with 265 additions and 50 deletions
--- a/machinery/src/cloud/cloud.go
+++ b/machinery/src/cloud/cloud.go
@@ -874,7 +874,7 @@ func HandleLiveStreamSD(livestreamCursor *packets.QueueCursor, configuration *mo
 	log.Log.Debug("cloud.HandleLiveStreamSD(): finished")
 }

-func HandleLiveStreamHD(livestreamCursor *packets.QueueCursor, configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, rtspClient capture.RTSPClient) {
+func HandleLiveStreamHD(configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, rtspClient capture.RTSPClient, rtspSubClient capture.RTSPClient, subStreamEnabled bool) {

 	config := configuration.Config

@@ -888,23 +888,51 @@ func HandleLiveStreamHD(livestreamCursor *packets.QueueCursor, configuration *mo
 			// Create per-peer broadcasters instead of shared tracks.
 			// Each viewer gets its own track with independent, non-blocking writes
 			// so a slow/congested peer cannot stall the others.
-			streams, _ := rtspClient.GetStreams()
-			videoBroadcaster := webrtc.NewVideoBroadcaster(streams)
-			audioBroadcaster := webrtc.NewAudioBroadcaster(streams)
+			//
+			// Both the main (high-resolution) and sub (low-resolution) streams are
+			// exposed as separate broadcasters that are always forwarding, so a
+			// viewer can pick the resolution it needs per peer connection without
+			// the agent re-negotiating the RTSP source.
+			mainStreams, _ := rtspClient.GetStreams()
+			mainVideoBroadcaster := webrtc.NewVideoBroadcaster(mainStreams)
+			mainAudioBroadcaster := webrtc.NewAudioBroadcaster(mainStreams)

-			if videoBroadcaster == nil && audioBroadcaster == nil {
-				log.Log.Error("cloud.HandleLiveStreamHD(): failed to create both video and audio broadcasters")
+			if mainVideoBroadcaster == nil && mainAudioBroadcaster == nil {
+				log.Log.Error("cloud.HandleLiveStreamHD(): failed to create both video and audio broadcasters for the main stream")
 				return
 			}

-			go webrtc.WriteToTrack(livestreamCursor, configuration, communication, mqttClient, videoBroadcaster, audioBroadcaster, rtspClient)
+			go webrtc.WriteToTrack(communication.Queue.Latest(), configuration, communication, mqttClient, mainVideoBroadcaster, mainAudioBroadcaster, rtspClient)
+
+			// Sub stream broadcasters, only when a distinct sub stream is available.
+			var subVideoBroadcaster *webrtc.TrackBroadcaster
+			var subAudioBroadcaster *webrtc.TrackBroadcaster
+			if subStreamEnabled && rtspSubClient != nil && communication.SubQueue != nil {
+				subStreams, _ := rtspSubClient.GetStreams()
+				subVideoBroadcaster = webrtc.NewVideoBroadcaster(subStreams)
+				subAudioBroadcaster = webrtc.NewAudioBroadcaster(subStreams)
+				go webrtc.WriteToTrack(communication.SubQueue.Latest(), configuration, communication, mqttClient, subVideoBroadcaster, subAudioBroadcaster, rtspSubClient)
+			}
+			subBroadcastersReady := subVideoBroadcaster != nil || subAudioBroadcaster != nil

 			if config.Capture.ForwardWebRTC == "true" {

 			} else {
 				log.Log.Info("cloud.HandleLiveStreamHD(): Waiting for peer connections.")
 				for handshake := range communication.HandleLiveHDHandshake {
-					log.Log.Info("cloud.HandleLiveStreamHD(): setting up a peer connection.")
+					// Route each viewer to the main or sub broadcasters based on the
+					// quality it requested; "auto" prefers the sub stream when one is
+					// available, matching the historical default.
+					useSub := models.SelectSubStreamForQuality(config, handshake.Payload.Quality, subStreamEnabled && subBroadcastersReady)
+					videoBroadcaster := mainVideoBroadcaster
+					audioBroadcaster := mainAudioBroadcaster
+					streamLabel := "main"
+					if useSub {
+						videoBroadcaster = subVideoBroadcaster
+						audioBroadcaster = subAudioBroadcaster
+						streamLabel = "sub"
+					}
+					log.Log.Info("cloud.HandleLiveStreamHD(): setting up a peer connection on the " + streamLabel + " stream (quality=" + handshake.Payload.Quality + ").")
 					go webrtc.InitializeWebRTCConnection(configuration, communication, mqttClient, videoBroadcaster, audioBroadcaster, handshake)
 				}
 			}
--- a/machinery/src/cloud/livehls.go
+++ b/machinery/src/cloud/livehls.go
@@ -6,7 +6,6 @@ import (

 	mqtt "github.com/eclipse/paho.mqtt.golang"

-	"github.com/kerberos-io/agent/machinery/src/capture"
 	"github.com/kerberos-io/agent/machinery/src/cloud/livehls"
 	"github.com/kerberos-io/agent/machinery/src/log"
 	"github.com/kerberos-io/agent/machinery/src/models"
@@ -46,7 +45,7 @@ const hlsReadyReannounceSeconds = 2
 // trades a little idle CPU for a near-instant "requesting stream", so viewers no
 // longer wait a full GOP for the first segment to be cut. Set
 // AGENT_LIVE_HLS_PREWARM=false to fall back to the lazy on-demand path above.
-func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, _ capture.RTSPClient) {
+func HandleLiveStreamHLS(configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, subStreamEnabled bool) {

 	log.Log.Debug("cloud.HandleLiveStreamHLS(): started")

@@ -81,10 +80,16 @@ func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *m
 		DeviceKey:     deviceId,
 	})

+	// The live session can be served from the main (high-resolution) or sub
+	// (low-resolution) stream and switched on demand. requestedQuality tracks the
+	// latest tier asked for over the keepalive; source holds the cursor plus the
+	// encoded parameter sets/dimensions for the stream currently being muxed.
 	// Encoded dimensions are only needed for the avcC fallback path (an SPS that
-	// mp4ff's strict parser rejects); the main stream dimensions are a safe value.
-	width := uint16(config.Capture.IPCamera.Width)
-	height := uint16(config.Capture.IPCamera.Height)
+	// mp4ff's strict parser rejects).
+	requestedQuality := models.StreamQualityAuto
+	useSub := models.SelectSubStreamForQuality(config, requestedQuality, subStreamEnabled)
+	source := buildHLSSource(config, communication, useSub)
+	log.Log.Info("cloud.HandleLiveStreamHLS(): serving live HLS from the " + source.label + " stream")

 	// prewarm keeps a single long-lived session muxing into an in-memory ring
 	// buffer while idle and flushes it the instant a viewer arrives, eliminating
@@ -117,12 +122,15 @@ func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *m
 	var pkt packets.Packet

 	for cursorError == nil {
-		pkt, cursorError = livestreamCursor.ReadPacket()
+		pkt, cursorError = source.cursor.ReadPacket()

 		now := time.Now().Unix()
 		select {
-		case <-communication.HandleLiveHLS:
+		case q := <-communication.HandleLiveHLS:
 			lastViewerRequest = now
+			if q != "" {
+				requestedQuality = q
+			}
 			// A keepalive may come from a viewer that just connected or hard-
 			// refreshed and therefore missed the one-shot readiness announcement
 			// fired when this session's first segment landed. Re-announce (throttled)
@@ -138,6 +146,22 @@ func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *m
 		default:
 		}

+		// Switch the source stream when the requested quality now maps to the other
+		// stream. Tearing the current session down makes the producer rebuild the
+		// init segment and announce a fresh session id from the new stream, which the
+		// viewer re-attaches to.
+		if wantSub := models.SelectSubStreamForQuality(config, requestedQuality, subStreamEnabled); wantSub != useSub {
+			useSub = wantSub
+			if session != nil {
+				_ = session.Close()
+				session = nil
+			}
+			source = buildHLSSource(config, communication, useSub)
+			lastReadyAnnounce = 0
+			log.Log.Info("cloud.HandleLiveStreamHLS(): switched live HLS to the " + source.label + " stream (quality=" + requestedQuality + ")")
+			continue
+		}
+
 		viewerActive := now-lastViewerRequest <= hlsViewerTimeoutSeconds

 		if prewarm {
@@ -151,11 +175,11 @@ func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *m
 				}
 				session = livehls.NewSession(publisher, livehls.SessionOptions{
 					Codec:          pkt.Codec,
-					SPSNALUs:       config.Capture.IPCamera.SPSNALUs,
-					PPSNALUs:       config.Capture.IPCamera.PPSNALUs,
-					VPSNALUs:       config.Capture.IPCamera.VPSNALUs,
-					Width:          width,
-					Height:         height,
+					SPSNALUs:       source.sps,
+					PPSNALUs:       source.pps,
+					VPSNALUs:       source.vps,
+					Width:          source.width,
+					Height:         source.height,
 					PartTargetMs:   partTargetMs,
 					StartBuffering: true,
 				})
@@ -211,11 +235,11 @@ func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *m
 			}
 			session = livehls.NewSession(publisher, livehls.SessionOptions{
 				Codec:        pkt.Codec,
-				SPSNALUs:     config.Capture.IPCamera.SPSNALUs,
-				PPSNALUs:     config.Capture.IPCamera.PPSNALUs,
-				VPSNALUs:     config.Capture.IPCamera.VPSNALUs,
-				Width:        width,
-				Height:       height,
+				SPSNALUs:     source.sps,
+				PPSNALUs:     source.pps,
+				VPSNALUs:     source.vps,
+				Width:        source.width,
+				Height:       source.height,
 				PartTargetMs: partTargetMs,
 			})
 			session.SetOnReady(func(sessionID string) {
@@ -259,3 +283,45 @@ func publishHLSReady(configuration *models.Configuration, mqttClient mqtt.Client
 		log.Log.Error("cloud.HandleLiveStreamHLS(): failed to package receive-hls-ready message: " + err.Error())
 	}
 }
+
+// hlsStreamSource bundles everything the live HLS producer needs to mux one of
+// the camera's streams: the packet cursor it reads from plus the encoded
+// parameter sets and dimensions used to build that stream's init segment.
+type hlsStreamSource struct {
+	cursor *packets.QueueCursor
+	sps    [][]byte
+	pps    [][]byte
+	vps    [][]byte
+	width  uint16
+	height uint16
+	label  string
+}
+
+// buildHLSSource resolves the packet cursor and encoded parameter sets/dimensions
+// for the selected stream. useSub picks the sub (low-resolution) stream when one
+// is available; otherwise the main (high-resolution) stream is used. A fresh
+// Latest() cursor is created so muxing resumes from the live edge of the chosen
+// stream after a switch.
+func buildHLSSource(config models.Config, communication *models.Communication, useSub bool) hlsStreamSource {
+	cam := config.Capture.IPCamera
+	if useSub && communication.SubQueue != nil {
+		return hlsStreamSource{
+			cursor: communication.SubQueue.Latest(),
+			sps:    cam.SubSPSNALUs,
+			pps:    cam.SubPPSNALUs,
+			vps:    cam.SubVPSNALUs,
+			width:  uint16(cam.SubWidth),
+			height: uint16(cam.SubHeight),
+			label:  "sub",
+		}
+	}
+	return hlsStreamSource{
+		cursor: communication.Queue.Latest(),
+		sps:    cam.SPSNALUs,
+		pps:    cam.PPSNALUs,
+		vps:    cam.VPSNALUs,
+		width:  uint16(cam.Width),
+		height: uint16(cam.Height),
+		label:  "main",
+	}
+}
--- a/machinery/src/components/kerberos.go
+++ b/machinery/src/components/kerberos.go
@@ -73,7 +73,7 @@ func Bootstrap(ctx context.Context, configDirectory string, configuration *model
 	communication.HandleLiveSDHTTP = make(chan int64, 1)
 	communication.HandleLiveHDKeepalive = make(chan string, 1)
 	communication.HandleLiveHDPeers = make(chan string, 1)
-	communication.HandleLiveHLS = make(chan int64, 1)
+	communication.HandleLiveHLS = make(chan string, 1)
 	communication.IsConfiguring = abool.New()

 	cameraSettings := &models.Camera{}
@@ -237,6 +237,13 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu
 		configuration.Config.Capture.IPCamera.SubWidth = width
 		configuration.Config.Capture.IPCamera.SubHeight = height

+		// Capture the sub stream parameter sets separately from the main stream so
+		// the live HLS muxer can build a correct init segment when a viewer asks for
+		// the sub (low-resolution) stream on demand.
+		configuration.Config.Capture.IPCamera.SubSPSNALUs = [][]byte{videoSubStream.SPS}
+		configuration.Config.Capture.IPCamera.SubPPSNALUs = [][]byte{videoSubStream.PPS}
+		configuration.Config.Capture.IPCamera.SubVPSNALUs = [][]byte{videoSubStream.VPS}
+
 		// If we have a substream, we need to set the width and height of the substream. (so we will override above information)
 		// Set the liveview width and height, this is used for the liveview and motion regions (drawing on the hub).
 		configuration.Config.Capture.IPCamera.BaseWidth, configuration.Config.Capture.IPCamera.BaseHeight =
@@ -287,26 +294,19 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu
 	}

 	// Handle livestream HLS (adaptive segments over HTTP via hub-api -> vault).
-	// Uses the sub stream when available (lower bitrate, browser-friendly), else
-	// the main stream. Like SD it is viewer-keepalive gated and produces no
-	// traffic while nobody is watching.
-	if subStreamEnabled {
-		livestreamHLSCursor := subQueue.Latest()
-		go cloud.HandleLiveStreamHLS(livestreamHLSCursor, configuration, communication, mqttClient, rtspSubClient)
-	} else {
-		livestreamHLSCursor := queue.Latest()
-		go cloud.HandleLiveStreamHLS(livestreamHLSCursor, configuration, communication, mqttClient, rtspClient)
-	}
+	// The producer can serve either the main (high-resolution) or sub
+	// (low-resolution) stream and switches between them on demand based on the
+	// quality the viewer requests; "auto" prefers the sub stream when available.
+	// Like SD it is viewer-keepalive gated and produces no traffic while nobody is
+	// watching.
+	go cloud.HandleLiveStreamHLS(configuration, communication, mqttClient, subStreamEnabled)

-	// Handle livestream HD (high resolution over WEBRTC)
+	// Handle livestream HD (high resolution over WEBRTC). Both the main and sub
+	// stream are exposed as separate broadcasters so a viewer can request the
+	// high (main) or low (sub) resolution per peer connection; "auto" prefers the
+	// sub stream when available.
 	communication.HandleLiveHDHandshake = make(chan models.LiveHDHandshake, 100)
-	if subStreamEnabled {
-		livestreamHDCursor := subQueue.Latest()
-		go cloud.HandleLiveStreamHD(livestreamHDCursor, configuration, communication, mqttClient, rtspSubClient)
-	} else {
-		livestreamHDCursor := queue.Latest()
-		go cloud.HandleLiveStreamHD(livestreamHDCursor, configuration, communication, mqttClient, rtspClient)
-	}
+	go cloud.HandleLiveStreamHD(configuration, communication, mqttClient, rtspClient, rtspSubClient, subStreamEnabled)

 	// Handle recording, will write an mp4 to disk.
 	go capture.HandleRecordStream(queue, configDirectory, configuration, communication, rtspClient)
--- a/machinery/src/models/communication.go
+++ b/machinery/src/models/communication.go
@@ -41,7 +41,10 @@ type Communication struct {
 	HandleLiveHDKeepalive chan string
 	HandleLiveHDHandshake chan LiveHDHandshake
 	HandleLiveHDPeers     chan string
-	HandleLiveHLS         chan int64
+	// HandleLiveHLS is the live HLS viewer keepalive. It carries the requested
+	// quality tier ("auto"|"high"|"low"; empty => auto) so the producer can switch
+	// the live session between the main and sub stream on demand.
+	HandleLiveHLS chan string
 	HandleONVIF           chan OnvifAction
 	IsConfiguring         *abool.AtomicBool
 	Queue                 *packets.Queue
--- a/machinery/src/models/config.go
+++ b/machinery/src/models/config.go
@@ -99,8 +99,14 @@ type IPCamera struct {
 	SPSNALUs      [][]byte `json:"sps_nalus,omitempty" bson:"sps_nalus,omitempty"`
 	PPSNALUs      [][]byte `json:"pps_nalus,omitempty" bson:"pps_nalus,omitempty"`
 	VPSNALUs      [][]byte `json:"vps_nalus,omitempty" bson:"vps_nalus,omitempty"`
-	SampleRate    int      `json:"sample_rate,omitempty" bson:"sample_rate,omitempty"`
-	Channels      int      `json:"channels,omitempty" bson:"channels,omitempty"`
+	// Sub stream parameter sets, captured separately from the main stream so the
+	// live HLS muxer can build a correct init segment when a viewer switches the
+	// live view to the sub (low-resolution) stream.
+	SubSPSNALUs [][]byte `json:"sub_sps_nalus,omitempty" bson:"sub_sps_nalus,omitempty"`
+	SubPPSNALUs [][]byte `json:"sub_pps_nalus,omitempty" bson:"sub_pps_nalus,omitempty"`
+	SubVPSNALUs [][]byte `json:"sub_vps_nalus,omitempty" bson:"sub_vps_nalus,omitempty"`
+	SampleRate  int      `json:"sample_rate,omitempty" bson:"sample_rate,omitempty"`
+	Channels    int      `json:"channels,omitempty" bson:"channels,omitempty"`
 }

 // USBCamera configuration, such as the device path (/dev/video*)
--- a/machinery/src/models/mqtt.go
+++ b/machinery/src/models/mqtt.go
@@ -179,11 +179,26 @@ type RequestSDStreamPayload struct {
 	Transport string `json:"transport,omitempty"`
 }

+// Stream quality tiers a viewer can request for the live (HD) view. The agent
+// maps these onto the camera's main (high-resolution) or sub (low-resolution)
+// RTSP stream, so a viewer can pick the resolution it needs instead of the agent
+// always preferring the sub stream. Empty/unknown values are treated as "auto"
+// for backward compatibility: older frontends that never set a quality keep the
+// previous behaviour (sub stream when available, otherwise main).
+const (
+	StreamQualityAuto = "auto" // agent decides based on availability/resolution
+	StreamQualityHigh = "high" // main stream (highest resolution)
+	StreamQualityLow  = "low"  // sub stream (lowest resolution)
+)
+
 // We received a live HLS stream request. Like SD it is a simple viewer
 // keepalive: the agent owns the live HLS session, so the request only needs to
-// signal "a viewer is watching" to keep the segment pipeline alive.
+// signal "a viewer is watching" to keep the segment pipeline alive. Quality lets
+// the viewer ask for the main (high) or sub (low) stream on demand; the agent
+// switches the live session's source stream when it changes.
 type RequestHLSStreamPayload struct {
-	Timestamp int64 `json:"timestamp"` // timestamp
+	Timestamp int64  `json:"timestamp"`         // timestamp
+	Quality   string `json:"quality,omitempty"` // "auto" | "high" | "low" (empty => auto)
 }

 // We received a request HD stream request
@@ -192,6 +207,7 @@ type RequestHDStreamPayload struct {
 	HubKey             string `json:"hub_key"`             // hub key
 	SessionID          string `json:"session_id"`          // session id
 	SessionDescription string `json:"session_description"` // session description
+	Quality            string `json:"quality,omitempty"`   // "auto" | "high" | "low" (empty => auto)
 }

 // We received a receive HD candidates request
--- a/machinery/src/models/stream.go
+++ b/machinery/src/models/stream.go
@@ -0,0 +1,40 @@
+package models
+
+// SelectSubStreamForQuality decides whether the live (HD) view should be served
+// from the sub (secondary) RTSP stream for the requested quality tier.
+//
+// It is resolution-aware: "high" picks whichever configured stream has the
+// higher resolution and "low" whichever has the lower resolution, regardless of
+// which one is wired as main vs sub. "auto" — the default, also used for the
+// empty/unknown value sent by older frontends that never set a quality — keeps
+// the historical behaviour of preferring the sub stream when one is available
+// (lower bitrate, browser friendly), falling back to the main stream otherwise.
+//
+// When no sub stream is configured the main stream is always used.
+func SelectSubStreamForQuality(config Config, quality string, subStreamEnabled bool) bool {
+	if !subStreamEnabled {
+		return false
+	}
+
+	cam := config.Capture.IPCamera
+	mainPixels := cam.Width * cam.Height
+	subPixels := cam.SubWidth * cam.SubHeight
+
+	switch quality {
+	case StreamQualityHigh:
+		// Highest resolution available. If the sub stream is (unusually) larger,
+		// use it; otherwise use the main stream. When dimensions are not yet known
+		// (0), default to the main stream for "high".
+		return subPixels > mainPixels
+	case StreamQualityLow:
+		// Lowest resolution available. If the main stream is (unusually) the
+		// smaller of the two, use it; otherwise use the sub stream. When the sub
+		// dimensions are unknown, still prefer the sub stream for "low".
+		if mainPixels > 0 && subPixels > 0 && mainPixels < subPixels {
+			return false
+		}
+		return true
+	default: // StreamQualityAuto, empty, or any unknown value
+		return true
+	}
+}
--- a/machinery/src/models/stream_test.go
+++ b/machinery/src/models/stream_test.go
@@ -0,0 +1,53 @@
+package models
+
+import "testing"
+
+func cfgWithDims(mainW, mainH, subW, subH int) Config {
+	c := Config{}
+	c.Capture.IPCamera.Width = mainW
+	c.Capture.IPCamera.Height = mainH
+	c.Capture.IPCamera.SubWidth = subW
+	c.Capture.IPCamera.SubHeight = subH
+	return c
+}
+
+func TestSelectSubStreamForQuality(t *testing.T) {
+	tests := []struct {
+		name             string
+		config           Config
+		quality          string
+		subStreamEnabled bool
+		wantSub          bool
+	}{
+		// No sub stream configured -> always the main stream.
+		{"no sub, auto", cfgWithDims(1920, 1080, 0, 0), StreamQualityAuto, false, false},
+		{"no sub, high", cfgWithDims(1920, 1080, 0, 0), StreamQualityHigh, false, false},
+		{"no sub, low", cfgWithDims(1920, 1080, 0, 0), StreamQualityLow, false, false},
+
+		// Typical config: main is the bigger stream, sub the smaller one.
+		{"auto prefers sub", cfgWithDims(1920, 1080, 640, 480), StreamQualityAuto, true, true},
+		{"empty prefers sub", cfgWithDims(1920, 1080, 640, 480), "", true, true},
+		{"unknown prefers sub", cfgWithDims(1920, 1080, 640, 480), "potato", true, true},
+		{"high picks main", cfgWithDims(1920, 1080, 640, 480), StreamQualityHigh, true, false},
+		{"low picks sub", cfgWithDims(1920, 1080, 640, 480), StreamQualityLow, true, true},
+
+		// Dimensions not probed yet (0): high defaults to main, low/auto to sub.
+		{"unknown dims, high", cfgWithDims(0, 0, 0, 0), StreamQualityHigh, true, false},
+		{"unknown dims, low", cfgWithDims(0, 0, 0, 0), StreamQualityLow, true, true},
+		{"unknown dims, auto", cfgWithDims(0, 0, 0, 0), StreamQualityAuto, true, true},
+
+		// Inverted config: sub is (unusually) the higher-resolution stream.
+		{"inverted high picks sub", cfgWithDims(640, 480, 1920, 1080), StreamQualityHigh, true, true},
+		{"inverted low picks main", cfgWithDims(640, 480, 1920, 1080), StreamQualityLow, true, false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := SelectSubStreamForQuality(tt.config, tt.quality, tt.subStreamEnabled)
+			if got != tt.wantSub {
+				t.Errorf("SelectSubStreamForQuality(quality=%q, subEnabled=%v) = %v, want %v",
+					tt.quality, tt.subStreamEnabled, got, tt.wantSub)
+			}
+		})
+	}
+}
--- a/machinery/src/routers/mqtt/main.go
+++ b/machinery/src/routers/mqtt/main.go
@@ -585,8 +585,11 @@ func HandleRequestHLSStream(mqttClient mqtt.Client, hubKey string, payload model

 	if requestHLSStreamPayload.Timestamp != 0 {
 		if communication.CameraConnected {
+			// Forward the requested quality ("auto"|"high"|"low"; empty => auto) so
+			// the producer can switch the live session between the main and sub
+			// stream on demand. The send doubles as the viewer keepalive.
 			select {
-			case communication.HandleLiveHLS <- time.Now().Unix():
+			case communication.HandleLiveHLS <- requestHLSStreamPayload.Quality:
 			default:
 			}
 			log.Log.Info("routers.mqtt.main.HandleRequestHLSStream(): received request to livestream over HLS.")