mirror of
https://github.com/kerberos-io/agent.git
synced 2026-08-23 15:08:32 +00:00
Merge pull request #290 from kerberos-io/feature/add-hls-live-streaming
feature/add-hls-live-streaming
This commit is contained in:
Cédric Verstraeten
177
machinery/src/cloud/livehls.go
Normal file
177
machinery/src/cloud/livehls.go
Normal file
@@ -0,0 +1,177 @@
|
||||
package cloud
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
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"
|
||||
"github.com/kerberos-io/agent/machinery/src/packets"
|
||||
)
|
||||
|
||||
// hlsViewerTimeoutSeconds is how long the agent keeps shipping live HLS segments
|
||||
// after the last viewer keepalive. It is a few seconds longer than the segment
|
||||
// duration so a viewer whose keepalive is briefly delayed does not cause the
|
||||
// session to flap. When it lapses the session is torn down to stop wasting
|
||||
// upload bandwidth when nobody is watching.
|
||||
const hlsViewerTimeoutSeconds = 8
|
||||
|
||||
// hlsReadyReannounceSeconds throttles how often the agent re-announces an
|
||||
// already-ready session over MQTT in response to viewer keepalives. The initial
|
||||
// "receive-hls-ready" is a one-shot fired when the first segment lands; a viewer
|
||||
// that connects or hard-refreshes after that (while the session is still alive)
|
||||
// missed it, so we re-announce on subsequent keepalives. Viewers dedupe by
|
||||
// session id, so a re-announce for a session they already play is a no-op. ~2s
|
||||
// gets a refreshed viewer playing well within its connection timeout without
|
||||
// spamming the control plane.
|
||||
const hlsReadyReannounceSeconds = 2
|
||||
|
||||
|
||||
// HandleLiveStreamHLS drives the live HLS producer. It mirrors HandleLiveStreamSD:
|
||||
// it reads the camera's packet stream from a Latest() cursor, and while a viewer
|
||||
// is active (kept alive via communication.HandleLiveHLS) it muxes the packets
|
||||
// into CMAF segments and ships them to hub-api, which stores each segment in an
|
||||
// ephemeral, short-TTL live window and serves the rolling playlist to viewers.
|
||||
//
|
||||
// A session is created lazily on the first keyframe seen while a viewer is active
|
||||
// and torn down once viewers go away, so an idle camera produces no live traffic.
|
||||
func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, _ capture.RTSPClient) {
|
||||
|
||||
log.Log.Debug("cloud.HandleLiveStreamHLS(): started")
|
||||
|
||||
config := configuration.Config
|
||||
|
||||
if config.Offline == "true" {
|
||||
log.Log.Debug("cloud.HandleLiveStreamHLS(): stopping as Offline is enabled.")
|
||||
return
|
||||
}
|
||||
if config.Capture.Liveview == "false" {
|
||||
log.Log.Debug("cloud.HandleLiveStreamHLS(): stopping as Liveview is disabled.")
|
||||
return
|
||||
}
|
||||
if config.HubURI == "" || config.HubKey == "" {
|
||||
log.Log.Debug("cloud.HandleLiveStreamHLS(): stopping as the Hub is not configured (HubURI/HubKey).")
|
||||
return
|
||||
}
|
||||
|
||||
hubKey := config.HubKey
|
||||
deviceId := config.Key
|
||||
|
||||
region := ""
|
||||
if config.S3 != nil {
|
||||
region = config.S3.Region
|
||||
}
|
||||
|
||||
publisher := livehls.NewPublisher(livehls.PublisherConfig{
|
||||
HubURI: config.HubURI,
|
||||
HubKey: config.HubKey,
|
||||
HubPrivateKey: config.HubPrivateKey,
|
||||
Region: region,
|
||||
DeviceKey: deviceId,
|
||||
})
|
||||
|
||||
// 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)
|
||||
|
||||
var session *livehls.Session
|
||||
lastViewerRequest := int64(0)
|
||||
lastReadyAnnounce := int64(0)
|
||||
|
||||
var cursorError error
|
||||
var pkt packets.Packet
|
||||
|
||||
for cursorError == nil {
|
||||
pkt, cursorError = livestreamCursor.ReadPacket()
|
||||
|
||||
now := time.Now().Unix()
|
||||
select {
|
||||
case <-communication.HandleLiveHLS:
|
||||
lastViewerRequest = now
|
||||
// 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)
|
||||
// so late/refreshed viewers learn the active session id; the frontend
|
||||
// dedupes by session id, so this is a no-op for viewers already playing.
|
||||
if session != nil && session.IsReady() && now-lastReadyAnnounce >= hlsReadyReannounceSeconds {
|
||||
publishHLSReady(configuration, mqttClient, hubKey, deviceId, session.SessionID())
|
||||
lastReadyAnnounce = now
|
||||
}
|
||||
default:
|
||||
}
|
||||
|
||||
viewerActive := now-lastViewerRequest <= hlsViewerTimeoutSeconds
|
||||
|
||||
if !viewerActive {
|
||||
// No viewer: stop and discard the session so we stop shipping segments.
|
||||
if session != nil {
|
||||
_ = session.Close()
|
||||
log.Log.Info("cloud.HandleLiveStreamHLS(): no active viewers, stopped live HLS session " + session.SessionID())
|
||||
session = nil
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
if len(pkt.Data) == 0 || !pkt.IsVideo {
|
||||
continue
|
||||
}
|
||||
|
||||
// Start a session lazily, but only on a keyframe so the first segment opens
|
||||
// on a random-access point.
|
||||
if session == nil {
|
||||
if !pkt.IsKeyFrame {
|
||||
continue
|
||||
}
|
||||
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,
|
||||
})
|
||||
session.SetOnReady(func(sessionID string) {
|
||||
log.Log.Info("cloud.HandleLiveStreamHLS(): live HLS session ready, announcing " + sessionID)
|
||||
publishHLSReady(configuration, mqttClient, hubKey, deviceId, sessionID)
|
||||
lastReadyAnnounce = time.Now().Unix()
|
||||
})
|
||||
log.Log.Info("cloud.HandleLiveStreamHLS(): started live HLS session " + session.SessionID())
|
||||
}
|
||||
|
||||
if err := session.WritePacket(pkt); err != nil {
|
||||
log.Log.Error("cloud.HandleLiveStreamHLS(): " + err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
if session != nil {
|
||||
_ = session.Close()
|
||||
}
|
||||
log.Log.Debug("cloud.HandleLiveStreamHLS(): finished")
|
||||
}
|
||||
|
||||
// publishHLSReady announces, over MQTT, that a live HLS session is available so
|
||||
// viewers can load the rolling playlist hub-api serves for {device}/{session}.
|
||||
func publishHLSReady(configuration *models.Configuration, mqttClient mqtt.Client, hubKey, deviceId, sessionID string) {
|
||||
valueMap := map[string]interface{}{
|
||||
"session": sessionID,
|
||||
"device": deviceId,
|
||||
}
|
||||
message := models.Message{
|
||||
Payload: models.Payload{
|
||||
Action: "receive-hls-ready",
|
||||
DeviceId: deviceId,
|
||||
Value: valueMap,
|
||||
},
|
||||
}
|
||||
payload, err := models.PackageMQTTMessage(configuration, message)
|
||||
if err == nil {
|
||||
mqttClient.Publish("kerberos/hub/"+hubKey, 0, false, payload)
|
||||
log.Log.Info("cloud.HandleLiveStreamHLS(): announced live HLS session " + sessionID)
|
||||
} else {
|
||||
log.Log.Error("cloud.HandleLiveStreamHLS(): failed to package receive-hls-ready message: " + err.Error())
|
||||
}
|
||||
}
|
||||
202
machinery/src/cloud/livehls/publisher.go
Normal file
202
machinery/src/cloud/livehls/publisher.go
Normal file
@@ -0,0 +1,202 @@
|
||||
// Package livehls implements the agent-side producer for live HLS streaming.
|
||||
//
|
||||
// It complements the recording pipeline: where recordings are muxed into one
|
||||
// fragmented MP4 and uploaded resumably (TUS) when complete, live HLS ships a
|
||||
// continuous series of small, independently-decodable CMAF segments to hub-api
|
||||
// the instant each is produced, so a browser can play a near-live HLS stream
|
||||
// without WebRTC/TURN (outbound HTTPS only).
|
||||
//
|
||||
// The wire contract (agent -> hub-api) intentionally mirrors the existing
|
||||
// header-based storage convention (X-Kerberos-Storage-Device / -FileName, plus
|
||||
// the Hub public/private key auth headers). hub-api authenticates the agent and
|
||||
// stores each segment in an ephemeral, short-TTL live window keyed by
|
||||
// {device}/{session}, which it serves straight back to the browser. The live
|
||||
// window is deliberately kept out of the vault and the recordings collection;
|
||||
// durable archival/DVR is a separate, later concern.
|
||||
//
|
||||
// Unlike recordings, live segments are NOT uploaded resumably: a 1-2s segment
|
||||
// that fails to upload is stale by the time a retry would land, so the publisher
|
||||
// is fire-and-forget and drops on failure (logged) rather than blocking the live
|
||||
// pipeline behind a retry/handshake.
|
||||
package livehls
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/kerberos-io/agent/machinery/src/log"
|
||||
"github.com/kerberos-io/agent/machinery/src/video"
|
||||
)
|
||||
|
||||
const (
|
||||
// liveIngestPath is the hub-api endpoint that accepts a single live segment
|
||||
// (or the init segment) and stores it in the ephemeral live window. hub-api
|
||||
// distinguishes init vs media segment and the object name via the
|
||||
// X-Kerberos-Live-* headers below, keeping a single route (mirrors the
|
||||
// existing /storage/upload convention).
|
||||
liveIngestPath = "/storage/live"
|
||||
|
||||
// Object names within a session. The init segment (ftyp+moov) is uploaded
|
||||
// once per session; media segments are seg-<sequence>.m4s.
|
||||
initObjectName = "init.mp4"
|
||||
|
||||
contentTypeInit = "video/mp4"
|
||||
contentTypeSegment = "video/iso.segment"
|
||||
|
||||
// Header names for the live ingest contract.
|
||||
headerHubPublicKey = "X-Kerberos-Hub-PublicKey"
|
||||
headerHubPrivateKey = "X-Kerberos-Hub-PrivateKey"
|
||||
headerHubRegion = "X-Kerberos-Hub-Region"
|
||||
headerStorageDevice = "X-Kerberos-Storage-Device"
|
||||
headerLiveSession = "X-Kerberos-Live-Session"
|
||||
headerLiveName = "X-Kerberos-Live-Name"
|
||||
headerLiveSequence = "X-Kerberos-Live-Sequence"
|
||||
headerLiveDuration = "X-Kerberos-Live-Duration"
|
||||
|
||||
// defaultPublishTimeout bounds a single segment upload. A live segment that
|
||||
// cannot be delivered within roughly its own duration is stale, so the upload
|
||||
// is abandoned (dropped) rather than allowed to back up the pipeline.
|
||||
defaultPublishTimeout = 4 * time.Second
|
||||
)
|
||||
|
||||
// PublisherConfig carries the hub endpoint and credentials needed to ship live
|
||||
// segments. It is populated from the agent's models.Config (HubURI/HubKey/...).
|
||||
type PublisherConfig struct {
|
||||
HubURI string // base hub-api URL, e.g. https://api.hub.example.com
|
||||
HubKey string // Hub public key (X-Kerberos-Hub-PublicKey)
|
||||
HubPrivateKey string // Hub private key (X-Kerberos-Hub-PrivateKey)
|
||||
Region string // storage region (X-Kerberos-Hub-Region), may be empty
|
||||
DeviceKey string // device/camera key (X-Kerberos-Storage-Device)
|
||||
|
||||
// Timeout optionally overrides defaultPublishTimeout (used by tests).
|
||||
Timeout time.Duration
|
||||
// HTTPClient optionally injects a client (used by tests). When nil a
|
||||
// redirect-credential-stripping client is created.
|
||||
HTTPClient *http.Client
|
||||
}
|
||||
|
||||
// Publisher ships init and media segments to hub-api over plain HTTP POST.
|
||||
//
|
||||
// It is safe for sequential use from a single live-stream goroutine. Methods are
|
||||
// fire-and-forget: they return an error for the caller to log, but the caller is
|
||||
// expected to continue (drop-on-fail) rather than retry.
|
||||
type Publisher struct {
|
||||
cfg PublisherConfig
|
||||
client *http.Client
|
||||
}
|
||||
|
||||
// NewPublisher builds a Publisher. The HTTP client strips the Hub credential
|
||||
// headers on a cross-host redirect (net/http does this for standard auth headers
|
||||
// but not custom-named ones), matching the recording upload client.
|
||||
func NewPublisher(cfg PublisherConfig) *Publisher {
|
||||
client := cfg.HTTPClient
|
||||
if client == nil {
|
||||
timeout := cfg.Timeout
|
||||
if timeout <= 0 {
|
||||
timeout = defaultPublishTimeout
|
||||
}
|
||||
client = &http.Client{
|
||||
Timeout: timeout,
|
||||
CheckRedirect: stripHubCredentialsOnCrossHostRedirect,
|
||||
}
|
||||
}
|
||||
return &Publisher{cfg: cfg, client: client}
|
||||
}
|
||||
|
||||
// PublishInit uploads the session's init segment (ftyp+moov). It must be called
|
||||
// (and succeed) before the player can use any media segment, so the caller
|
||||
// should treat a failure here as "session not yet established" and retry on the
|
||||
// next init opportunity rather than shipping media segments blindly.
|
||||
func (p *Publisher) PublishInit(ctx context.Context, sessionID string, data []byte) error {
|
||||
return p.post(ctx, postParams{
|
||||
sessionID: sessionID,
|
||||
name: initObjectName,
|
||||
contentType: contentTypeInit,
|
||||
body: data,
|
||||
})
|
||||
}
|
||||
|
||||
// PublishSegment uploads one media segment (styp+moof+mdat). The segment's
|
||||
// sequence number and duration travel in headers so hub-api can update the
|
||||
// rolling playlist window without parsing the box structure.
|
||||
func (p *Publisher) PublishSegment(ctx context.Context, sessionID string, seg video.LiveSegment) error {
|
||||
return p.post(ctx, postParams{
|
||||
sessionID: sessionID,
|
||||
name: fmt.Sprintf("seg-%d.m4s", seg.SequenceNumber),
|
||||
sequence: seg.SequenceNumber,
|
||||
durationMs: seg.DurationMs,
|
||||
hasSegment: true,
|
||||
contentType: contentTypeSegment,
|
||||
body: seg.Data,
|
||||
})
|
||||
}
|
||||
|
||||
type postParams struct {
|
||||
sessionID string
|
||||
name string
|
||||
sequence uint32
|
||||
durationMs uint64
|
||||
hasSegment bool
|
||||
contentType string
|
||||
body []byte
|
||||
}
|
||||
|
||||
// post performs a single fire-and-forget upload to the live ingest endpoint.
|
||||
func (p *Publisher) post(ctx context.Context, params postParams) error {
|
||||
if p.cfg.HubURI == "" {
|
||||
return fmt.Errorf("livehls: HubURI not configured")
|
||||
}
|
||||
if params.sessionID == "" {
|
||||
return fmt.Errorf("livehls: empty session id")
|
||||
}
|
||||
|
||||
url := strings.TrimRight(p.cfg.HubURI, "/") + liveIngestPath
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(params.body))
|
||||
if err != nil {
|
||||
return fmt.Errorf("livehls: build request: %w", err)
|
||||
}
|
||||
|
||||
req.Header.Set("Content-Type", params.contentType)
|
||||
req.Header.Set(headerStorageDevice, p.cfg.DeviceKey)
|
||||
req.Header.Set(headerLiveSession, params.sessionID)
|
||||
req.Header.Set(headerLiveName, params.name)
|
||||
if params.hasSegment {
|
||||
req.Header.Set(headerLiveSequence, strconv.FormatUint(uint64(params.sequence), 10))
|
||||
req.Header.Set(headerLiveDuration, strconv.FormatUint(params.durationMs, 10))
|
||||
}
|
||||
req.Header.Set(headerHubPublicKey, p.cfg.HubKey)
|
||||
req.Header.Set(headerHubPrivateKey, p.cfg.HubPrivateKey)
|
||||
req.Header.Set(headerHubRegion, p.cfg.Region)
|
||||
|
||||
resp, err := p.client.Do(req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("livehls: upload %s: %w", params.name, err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||
return fmt.Errorf("livehls: upload %s rejected: %s", params.name, resp.Status)
|
||||
}
|
||||
log.Log.Debug("livehls.Publisher.post(): shipped " + params.name + " for session " + params.sessionID)
|
||||
return nil
|
||||
}
|
||||
|
||||
// stripHubCredentialsOnCrossHostRedirect removes the Hub credential headers when
|
||||
// a redirect crosses to a different host. net/http strips standard sensitive
|
||||
// headers on a cross-host redirect but not custom-named ones, so without this the
|
||||
// Hub keys could leak to a redirect target.
|
||||
func stripHubCredentialsOnCrossHostRedirect(req *http.Request, via []*http.Request) error {
|
||||
if len(via) == 0 {
|
||||
return nil
|
||||
}
|
||||
if req.URL.Host != via[0].URL.Host {
|
||||
req.Header.Del(headerHubPrivateKey)
|
||||
req.Header.Del(headerHubPublicKey)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
312
machinery/src/cloud/livehls/publisher_test.go
Normal file
312
machinery/src/cloud/livehls/publisher_test.go
Normal file
@@ -0,0 +1,312 @@
|
||||
package livehls
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/kerberos-io/agent/machinery/src/packets"
|
||||
"github.com/kerberos-io/agent/machinery/src/video"
|
||||
)
|
||||
|
||||
// captured records one received upload for assertions.
|
||||
type captured struct {
|
||||
path string
|
||||
method string
|
||||
contentType string
|
||||
device string
|
||||
session string
|
||||
name string
|
||||
sequence string
|
||||
duration string
|
||||
hubPublic string
|
||||
hubPrivate string
|
||||
region string
|
||||
body []byte
|
||||
}
|
||||
|
||||
// newCapturingServer returns an httptest server that records every upload and
|
||||
// replies with the given status code.
|
||||
func newCapturingServer(t *testing.T, status int) (*httptest.Server, *[]captured, *sync.Mutex) {
|
||||
t.Helper()
|
||||
var mu sync.Mutex
|
||||
var got []captured
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
body, _ := io.ReadAll(r.Body)
|
||||
mu.Lock()
|
||||
got = append(got, captured{
|
||||
path: r.URL.Path,
|
||||
method: r.Method,
|
||||
contentType: r.Header.Get("Content-Type"),
|
||||
device: r.Header.Get(headerStorageDevice),
|
||||
session: r.Header.Get(headerLiveSession),
|
||||
name: r.Header.Get(headerLiveName),
|
||||
sequence: r.Header.Get(headerLiveSequence),
|
||||
duration: r.Header.Get(headerLiveDuration),
|
||||
hubPublic: r.Header.Get(headerHubPublicKey),
|
||||
hubPrivate: r.Header.Get(headerHubPrivateKey),
|
||||
region: r.Header.Get(headerHubRegion),
|
||||
body: body,
|
||||
})
|
||||
mu.Unlock()
|
||||
w.WriteHeader(status)
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
return srv, &got, &mu
|
||||
}
|
||||
|
||||
func testPublisher(hubURI string) *Publisher {
|
||||
return NewPublisher(PublisherConfig{
|
||||
HubURI: hubURI,
|
||||
HubKey: "pub-key",
|
||||
HubPrivateKey: "priv-key",
|
||||
Region: "eu-west",
|
||||
DeviceKey: "cam-1",
|
||||
Timeout: 2 * time.Second,
|
||||
})
|
||||
}
|
||||
|
||||
func TestPublisherPublishInitSendsContractHeaders(t *testing.T) {
|
||||
srv, got, mu := newCapturingServer(t, http.StatusOK)
|
||||
p := testPublisher(srv.URL)
|
||||
|
||||
if err := p.PublishInit(context.Background(), "sess-1", []byte("INITBYTES")); err != nil {
|
||||
t.Fatalf("PublishInit: %v", err)
|
||||
}
|
||||
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
if len(*got) != 1 {
|
||||
t.Fatalf("server received %d requests, want 1", len(*got))
|
||||
}
|
||||
c := (*got)[0]
|
||||
if c.method != http.MethodPost {
|
||||
t.Errorf("method=%s, want POST", c.method)
|
||||
}
|
||||
if c.path != liveIngestPath {
|
||||
t.Errorf("path=%s, want %s", c.path, liveIngestPath)
|
||||
}
|
||||
if c.contentType != contentTypeInit {
|
||||
t.Errorf("content-type=%s, want %s", c.contentType, contentTypeInit)
|
||||
}
|
||||
if c.device != "cam-1" {
|
||||
t.Errorf("device=%s, want cam-1", c.device)
|
||||
}
|
||||
if c.session != "sess-1" {
|
||||
t.Errorf("session=%s, want sess-1", c.session)
|
||||
}
|
||||
if c.name != initObjectName {
|
||||
t.Errorf("name=%s, want %s", c.name, initObjectName)
|
||||
}
|
||||
if c.hubPublic != "pub-key" || c.hubPrivate != "priv-key" || c.region != "eu-west" {
|
||||
t.Errorf("auth headers wrong: pub=%q priv=%q region=%q", c.hubPublic, c.hubPrivate, c.region)
|
||||
}
|
||||
if string(c.body) != "INITBYTES" {
|
||||
t.Errorf("body=%q, want INITBYTES", string(c.body))
|
||||
}
|
||||
// init must NOT carry segment-only headers.
|
||||
if c.sequence != "" || c.duration != "" {
|
||||
t.Errorf("init should not send sequence/duration, got seq=%q dur=%q", c.sequence, c.duration)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPublisherPublishSegmentSendsSequenceAndDuration(t *testing.T) {
|
||||
srv, got, mu := newCapturingServer(t, http.StatusOK)
|
||||
p := testPublisher(srv.URL)
|
||||
|
||||
seg := video.LiveSegment{SequenceNumber: 7, DurationMs: 1960, Data: []byte("SEGMENT")}
|
||||
if err := p.PublishSegment(context.Background(), "sess-9", seg); err != nil {
|
||||
t.Fatalf("PublishSegment: %v", err)
|
||||
}
|
||||
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
c := (*got)[0]
|
||||
if c.contentType != contentTypeSegment {
|
||||
t.Errorf("content-type=%s, want %s", c.contentType, contentTypeSegment)
|
||||
}
|
||||
if c.name != "seg-7.m4s" {
|
||||
t.Errorf("name=%s, want seg-7.m4s", c.name)
|
||||
}
|
||||
if c.sequence != "7" {
|
||||
t.Errorf("sequence=%s, want 7", c.sequence)
|
||||
}
|
||||
if c.duration != "1960" {
|
||||
t.Errorf("duration=%s, want 1960", c.duration)
|
||||
}
|
||||
if string(c.body) != "SEGMENT" {
|
||||
t.Errorf("body=%q, want SEGMENT", string(c.body))
|
||||
}
|
||||
}
|
||||
|
||||
func TestPublisherReturnsErrorOnNon2xx(t *testing.T) {
|
||||
srv, _, _ := newCapturingServer(t, http.StatusInternalServerError)
|
||||
p := testPublisher(srv.URL)
|
||||
|
||||
err := p.PublishSegment(context.Background(), "s", video.LiveSegment{SequenceNumber: 1, Data: []byte("x")})
|
||||
if err == nil {
|
||||
t.Fatal("expected an error on 500 response")
|
||||
}
|
||||
}
|
||||
|
||||
func TestPublisherErrorsWithoutHubURI(t *testing.T) {
|
||||
p := NewPublisher(PublisherConfig{DeviceKey: "cam"})
|
||||
if err := p.PublishInit(context.Background(), "s", []byte("x")); err == nil {
|
||||
t.Fatal("expected error when HubURI is empty")
|
||||
}
|
||||
}
|
||||
|
||||
// makeAnnexBVideoPacket builds a synthetic capture packet carrying one Annex B
|
||||
// H.264 access unit at the given decode time (ms).
|
||||
func makeAnnexBVideoPacket(isKey bool, timeMs int64) packets.Packet {
|
||||
nalType := byte(0x01)
|
||||
if isKey {
|
||||
nalType = 0x65
|
||||
}
|
||||
data := []byte{0x00, 0x00, 0x00, 0x01, nalType}
|
||||
for i := 0; i < 80; i++ {
|
||||
data = append(data, byte(i))
|
||||
}
|
||||
return packets.Packet{
|
||||
IsVideo: true,
|
||||
IsKeyFrame: isKey,
|
||||
Codec: "H264",
|
||||
Data: data,
|
||||
TimeLegacy: time.Duration(timeMs) * time.Millisecond,
|
||||
}
|
||||
}
|
||||
|
||||
func TestSessionShipsInitThenSegmentsAndFiresReady(t *testing.T) {
|
||||
srv, got, mu := newCapturingServer(t, http.StatusOK)
|
||||
p := testPublisher(srv.URL)
|
||||
|
||||
sess := NewSession(p, SessionOptions{
|
||||
Codec: "H264",
|
||||
SPSNALUs: [][]byte{liveTestSPSForSession()},
|
||||
PPSNALUs: [][]byte{{0x68, 0xce, 0x38, 0x80}},
|
||||
Width: 640,
|
||||
Height: 480,
|
||||
TargetSegmentMs: 2000,
|
||||
})
|
||||
|
||||
var readyCalls int
|
||||
var readySession string
|
||||
sess.SetOnReady(func(id string) {
|
||||
readyCalls++
|
||||
readySession = id
|
||||
})
|
||||
|
||||
// 4 GOPs of 25 frames @ 40ms = 1s GOPs => with 2s target, 2 segments emitted
|
||||
// during streaming and a final one on Close.
|
||||
const gopFrames, gops = 25, 4
|
||||
for i := 0; i < gopFrames*gops; i++ {
|
||||
isKey := i%gopFrames == 0
|
||||
pkt := makeAnnexBVideoPacket(isKey, int64(i*40))
|
||||
if err := sess.WritePacket(pkt); err != nil {
|
||||
t.Fatalf("WritePacket(%d): %v", i, err)
|
||||
}
|
||||
}
|
||||
// A non-video packet must be ignored.
|
||||
if err := sess.WritePacket(packets.Packet{IsAudio: true, Data: []byte{1, 2, 3}}); err != nil {
|
||||
t.Fatalf("WritePacket(audio): %v", err)
|
||||
}
|
||||
if err := sess.Close(); err != nil {
|
||||
t.Fatalf("Close: %v", err)
|
||||
}
|
||||
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
|
||||
var initCount, segCount int
|
||||
for _, c := range *got {
|
||||
if c.name == initObjectName {
|
||||
initCount++
|
||||
if string(c.body[4:8]) != "ftyp" {
|
||||
t.Errorf("init body is not an ftyp box: % x", c.body[:12])
|
||||
}
|
||||
} else {
|
||||
segCount++
|
||||
if c.session != sess.SessionID() {
|
||||
t.Errorf("segment session=%s, want %s", c.session, sess.SessionID())
|
||||
}
|
||||
}
|
||||
}
|
||||
if initCount != 1 {
|
||||
t.Errorf("init uploaded %d times, want exactly 1", initCount)
|
||||
}
|
||||
if segCount < 2 {
|
||||
t.Errorf("got %d segment uploads, want >= 2", segCount)
|
||||
}
|
||||
if readyCalls != 1 {
|
||||
t.Errorf("OnReady fired %d times, want exactly 1", readyCalls)
|
||||
}
|
||||
if readySession != sess.SessionID() {
|
||||
t.Errorf("OnReady session=%s, want %s", readySession, sess.SessionID())
|
||||
}
|
||||
}
|
||||
|
||||
func TestSessionRetriesInitWhenFirstAttemptFails(t *testing.T) {
|
||||
// Server fails the first N requests, then succeeds. This proves init is
|
||||
// re-attempted (not dropped) so the session can still establish.
|
||||
var mu sync.Mutex
|
||||
var inits, segs int
|
||||
failFirst := 1
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
name := r.Header.Get(headerLiveName)
|
||||
if name == initObjectName {
|
||||
inits++
|
||||
if inits <= failFirst {
|
||||
w.WriteHeader(http.StatusBadGateway)
|
||||
return
|
||||
}
|
||||
} else {
|
||||
segs++
|
||||
}
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
|
||||
sess := NewSession(testPublisher(srv.URL), SessionOptions{
|
||||
Codec: "H264",
|
||||
SPSNALUs: [][]byte{liveTestSPSForSession()},
|
||||
PPSNALUs: [][]byte{{0x68, 0xce, 0x38, 0x80}},
|
||||
Width: 640,
|
||||
Height: 480,
|
||||
})
|
||||
|
||||
var ready int
|
||||
sess.SetOnReady(func(string) { ready++ })
|
||||
|
||||
for i := 0; i < 60; i++ {
|
||||
isKey := i%25 == 0
|
||||
if err := sess.WritePacket(makeAnnexBVideoPacket(isKey, int64(i*40))); err != nil {
|
||||
t.Fatalf("WritePacket(%d): %v", i, err)
|
||||
}
|
||||
}
|
||||
if err := sess.Close(); err != nil {
|
||||
t.Fatalf("Close: %v", err)
|
||||
}
|
||||
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
if inits < 2 {
|
||||
t.Errorf("init attempted %d times, want >= 2 (first failed then retried)", inits)
|
||||
}
|
||||
if segs < 1 {
|
||||
t.Errorf("no segments delivered after init recovered (segs=%d)", segs)
|
||||
}
|
||||
if ready != 1 {
|
||||
t.Errorf("OnReady fired %d times, want 1", ready)
|
||||
}
|
||||
}
|
||||
|
||||
// liveTestSPSForSession is the known-good baseline SPS reused across tests.
|
||||
func liveTestSPSForSession() []byte {
|
||||
return []byte{0x67, 0x42, 0xc0, 0x1e, 0xd9, 0x00, 0xa0, 0x47, 0xfe, 0xc8}
|
||||
}
|
||||
255
machinery/src/cloud/livehls/session.go
Normal file
255
machinery/src/cloud/livehls/session.go
Normal file
@@ -0,0 +1,255 @@
|
||||
package livehls
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/kerberos-io/agent/machinery/src/log"
|
||||
"github.com/kerberos-io/agent/machinery/src/packets"
|
||||
"github.com/kerberos-io/agent/machinery/src/video"
|
||||
)
|
||||
|
||||
// DefaultTargetSegmentMs is the nominal live segment length. ~2s keeps standard
|
||||
// HLS latency reasonable (a player typically buffers ~3 segments) while staying
|
||||
// large enough that per-segment HTTP overhead is negligible.
|
||||
const DefaultTargetSegmentMs = 2000
|
||||
|
||||
// Session ties a video.LiveSegmenter to a Publisher: it converts capture packets
|
||||
// into CMAF segments and ships each one to hub-api. Exactly one init segment is
|
||||
// delivered per session (re-attempted until it lands), after which media
|
||||
// segments are published and the OnReady signal fires once so the control plane
|
||||
// (MQTT) can tell viewers the live playlist exists.
|
||||
//
|
||||
// A Session is driven from a single goroutine (the live-stream loop); its methods
|
||||
// are not safe for concurrent use except SessionID, which is immutable.
|
||||
type Session struct {
|
||||
id string
|
||||
publisher *Publisher
|
||||
segmenter *video.LiveSegmenter
|
||||
|
||||
// newContext produces the per-upload context (timeout). Overridable in tests.
|
||||
newContext func() (context.Context, context.CancelFunc)
|
||||
|
||||
mu sync.Mutex
|
||||
initBytes []byte
|
||||
initPublished bool
|
||||
// lastInitAt is when the init segment was last (re)uploaded. The init is
|
||||
// re-sent periodically so its short TTL in the hub live window never lapses
|
||||
// mid-session; see refreshInitIfStale.
|
||||
lastInitAt time.Time
|
||||
readyFired bool
|
||||
onReady func(sessionID string)
|
||||
}
|
||||
|
||||
// SessionOptions configures a live HLS session.
|
||||
type SessionOptions struct {
|
||||
Codec string // "H264" or "H265"
|
||||
SPSNALUs [][]byte // parameter sets (raw or Annex B)
|
||||
PPSNALUs [][]byte //
|
||||
VPSNALUs [][]byte // H.265 only
|
||||
Width uint16 // encoded width (for the avcC fallback path)
|
||||
Height uint16 // encoded height
|
||||
TargetSegmentMs uint64 // 0 => DefaultTargetSegmentMs
|
||||
}
|
||||
|
||||
// NewSession builds a session with a fresh random id and wires the segmenter's
|
||||
// init/segment callbacks to the publisher.
|
||||
func NewSession(publisher *Publisher, opts SessionOptions) *Session {
|
||||
target := opts.TargetSegmentMs
|
||||
if target == 0 {
|
||||
target = DefaultTargetSegmentMs
|
||||
}
|
||||
seg := video.NewLiveSegmenter(opts.Codec, opts.SPSNALUs, opts.PPSNALUs, opts.VPSNALUs, target)
|
||||
seg.SetDimensions(opts.Width, opts.Height)
|
||||
|
||||
s := &Session{
|
||||
id: newSessionID(),
|
||||
publisher: publisher,
|
||||
segmenter: seg,
|
||||
newContext: func() (context.Context, context.CancelFunc) {
|
||||
return context.WithTimeout(context.Background(), defaultPublishTimeout)
|
||||
},
|
||||
}
|
||||
|
||||
// The segmenter emits the init segment exactly once; capture it and try to
|
||||
// ship it. Failures here are non-fatal - publishInitIfNeeded re-attempts
|
||||
// before the next media segment so a transient hub hiccup at startup does not
|
||||
// permanently break the session.
|
||||
seg.OnInit = func(initBytes []byte) error {
|
||||
s.mu.Lock()
|
||||
s.initBytes = append([]byte(nil), initBytes...)
|
||||
s.mu.Unlock()
|
||||
s.publishInitIfNeeded()
|
||||
return nil
|
||||
}
|
||||
|
||||
// Each completed media segment is shipped. We only publish a segment once the
|
||||
// init segment has landed (a media segment is useless without it), and we fire
|
||||
// OnReady after the first successfully shipped segment.
|
||||
seg.OnSegment = func(segment video.LiveSegment) error {
|
||||
if !s.publishInitIfNeeded() {
|
||||
log.Log.Warning("livehls.Session: dropping segment " +
|
||||
fmt.Sprintf("%d", segment.SequenceNumber) + " because init has not been delivered yet")
|
||||
return nil
|
||||
}
|
||||
ctx, cancel := s.newContext()
|
||||
defer cancel()
|
||||
if err := s.publisher.PublishSegment(ctx, s.id, segment); err != nil {
|
||||
log.Log.Warning("livehls.Session: " + err.Error())
|
||||
return nil
|
||||
}
|
||||
s.fireReadyOnce()
|
||||
// Keep the (write-once) init segment from ageing out of the live window
|
||||
// while the session is still producing media.
|
||||
s.refreshInitIfStale()
|
||||
return nil
|
||||
}
|
||||
|
||||
return s
|
||||
}
|
||||
|
||||
// SessionID returns the immutable session identifier used in object keys and the
|
||||
// MQTT ready signal.
|
||||
func (s *Session) SessionID() string { return s.id }
|
||||
|
||||
// IsReady reports whether the session has delivered its init segment and at
|
||||
// least one media segment, i.e. the playlist hub-api serves is now playable. It
|
||||
// lets the live-stream loop re-announce "receive-hls-ready" to viewers that join
|
||||
// or hard-refresh after the initial one-shot signal (which they would otherwise
|
||||
// never receive, leaving the stream blank until the session is recreated).
|
||||
func (s *Session) IsReady() bool {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return s.readyFired
|
||||
}
|
||||
|
||||
// SetOnReady registers a callback fired exactly once, after the first media
|
||||
// segment has been successfully delivered. Used to publish the MQTT
|
||||
// "receive-hls-ready" signal so viewers can load the playlist.
|
||||
func (s *Session) SetOnReady(fn func(sessionID string)) {
|
||||
s.mu.Lock()
|
||||
s.onReady = fn
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
// WritePacket feeds one capture packet into the segmenter. Non-video packets are
|
||||
// ignored (the spike is video-only). The decode timestamp is derived exactly as
|
||||
// the recording muxer does: DTS = PTS - compositionOffset, with the composition
|
||||
// offset forwarded for correct B-frame presentation order.
|
||||
func (s *Session) WritePacket(pkt packets.Packet) error {
|
||||
if !pkt.IsVideo {
|
||||
return nil
|
||||
}
|
||||
pts := uint64(pkt.TimeLegacy.Milliseconds())
|
||||
compositionOffset := pkt.CompositionTime
|
||||
dts := pts
|
||||
if compositionOffset > 0 && uint64(compositionOffset) <= pts {
|
||||
dts = pts - uint64(compositionOffset)
|
||||
} else if compositionOffset < 0 || uint64(compositionOffset) > pts {
|
||||
// Guard against invalid offsets to avoid producing a CTS (DTS+CTO) jump.
|
||||
compositionOffset = 0
|
||||
}
|
||||
return s.segmenter.WriteSample(pkt.IsKeyFrame, pkt.Data, dts, int32(compositionOffset))
|
||||
}
|
||||
|
||||
// Close flushes any buffered sample and ships the final segment.
|
||||
func (s *Session) Close() error {
|
||||
return s.segmenter.Close()
|
||||
}
|
||||
|
||||
// publishInitIfNeeded ensures the init segment has been delivered, attempting an
|
||||
// upload if it has not. Returns true once init is known to be published.
|
||||
func (s *Session) publishInitIfNeeded() bool {
|
||||
s.mu.Lock()
|
||||
if s.initPublished {
|
||||
s.mu.Unlock()
|
||||
return true
|
||||
}
|
||||
initBytes := s.initBytes
|
||||
s.mu.Unlock()
|
||||
|
||||
if len(initBytes) == 0 {
|
||||
return false
|
||||
}
|
||||
|
||||
ctx, cancel := s.newContext()
|
||||
defer cancel()
|
||||
if err := s.publisher.PublishInit(ctx, s.id, initBytes); err != nil {
|
||||
log.Log.Warning("livehls.Session: init upload failed, will retry: " + err.Error())
|
||||
return false
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
s.initPublished = true
|
||||
s.lastInitAt = time.Now()
|
||||
s.mu.Unlock()
|
||||
log.Log.Info("livehls.Session: init segment delivered for session " + s.id)
|
||||
return true
|
||||
}
|
||||
|
||||
// initRefreshInterval is how often the init segment is re-uploaded so its TTL in
|
||||
// the hub-api live window never lapses mid-session. The init segment is otherwise
|
||||
// written only once per session; because the live window expires objects after a
|
||||
// short TTL (LiveSegmentTTLSeconds, 45s on the hub) the init would age out after
|
||||
// ~1 minute and the playlist's #EXT-X-MAP would start 404ing, stalling playback.
|
||||
// Re-uploading well inside that TTL keeps the init alive for the life of the
|
||||
// session while still letting it expire naturally once the session ends.
|
||||
const initRefreshInterval = 15 * time.Second
|
||||
|
||||
// refreshInitIfStale re-uploads the init segment if it has not been refreshed
|
||||
// within initRefreshInterval, keeping its created_at (and thus its TTL) current
|
||||
// for as long as the session is producing segments. It is a no-op until the init
|
||||
// has first been published. Failures are non-fatal: the next segment retries.
|
||||
func (s *Session) refreshInitIfStale() {
|
||||
s.mu.Lock()
|
||||
if !s.initPublished || time.Since(s.lastInitAt) < initRefreshInterval {
|
||||
s.mu.Unlock()
|
||||
return
|
||||
}
|
||||
initBytes := s.initBytes
|
||||
s.mu.Unlock()
|
||||
|
||||
if len(initBytes) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
ctx, cancel := s.newContext()
|
||||
defer cancel()
|
||||
if err := s.publisher.PublishInit(ctx, s.id, initBytes); err != nil {
|
||||
log.Log.Warning("livehls.Session: init refresh failed, will retry: " + err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
s.lastInitAt = time.Now()
|
||||
s.mu.Unlock()
|
||||
log.Log.Debug("livehls.Session: refreshed init segment TTL for session " + s.id)
|
||||
}
|
||||
|
||||
// fireReadyOnce invokes the OnReady callback the first time it is called.
|
||||
func (s *Session) fireReadyOnce() {
|
||||
s.mu.Lock()
|
||||
if s.readyFired || s.onReady == nil {
|
||||
s.mu.Unlock()
|
||||
return
|
||||
}
|
||||
s.readyFired = true
|
||||
fn := s.onReady
|
||||
s.mu.Unlock()
|
||||
fn(s.id)
|
||||
}
|
||||
|
||||
// newSessionID returns a short, unique, URL-safe session identifier of the form
|
||||
// <unix-seconds>-<random-hex>.
|
||||
func newSessionID() string {
|
||||
b := make([]byte, 4)
|
||||
if _, err := rand.Read(b); err != nil {
|
||||
// rand.Read essentially never fails; fall back to a time-only id.
|
||||
return fmt.Sprintf("%d", time.Now().UnixNano())
|
||||
}
|
||||
return fmt.Sprintf("%d-%s", time.Now().Unix(), hex.EncodeToString(b))
|
||||
}
|
||||
@@ -72,6 +72,7 @@ func Bootstrap(ctx context.Context, configDirectory string, configuration *model
|
||||
communication.HandleLiveSD = make(chan int64, 1)
|
||||
communication.HandleLiveHDKeepalive = make(chan string, 1)
|
||||
communication.HandleLiveHDPeers = make(chan string, 1)
|
||||
communication.HandleLiveHLS = make(chan int64, 1)
|
||||
communication.IsConfiguring = abool.New()
|
||||
|
||||
cameraSettings := &models.Camera{}
|
||||
@@ -304,6 +305,18 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu
|
||||
go cloud.HandleLiveStreamSD(livestreamCursor, configuration, communication, mqttClient, rtspClient)
|
||||
}
|
||||
|
||||
// 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)
|
||||
}
|
||||
|
||||
// Handle livestream HD (high resolution over WEBRTC)
|
||||
communication.HandleLiveHDHandshake = make(chan models.LiveHDHandshake, 100)
|
||||
if subStreamEnabled {
|
||||
|
||||
@@ -40,6 +40,7 @@ type Communication struct {
|
||||
HandleLiveHDKeepalive chan string
|
||||
HandleLiveHDHandshake chan LiveHDHandshake
|
||||
HandleLiveHDPeers chan string
|
||||
HandleLiveHLS chan int64
|
||||
HandleONVIF chan OnvifAction
|
||||
IsConfiguring *abool.AtomicBool
|
||||
Queue *packets.Queue
|
||||
|
||||
@@ -173,6 +173,13 @@ type RequestSDStreamPayload struct {
|
||||
Timestamp int64 `json:"timestamp"` // timestamp
|
||||
}
|
||||
|
||||
// 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.
|
||||
type RequestHLSStreamPayload struct {
|
||||
Timestamp int64 `json:"timestamp"` // timestamp
|
||||
}
|
||||
|
||||
// We received a request HD stream request
|
||||
type RequestHDStreamPayload struct {
|
||||
Timestamp int64 `json:"timestamp"` // timestamp
|
||||
|
||||
@@ -344,6 +344,8 @@ func MQTTListenerHandler(mqttClient mqtt.Client, hubKey string, configDirectory
|
||||
go HandleRequestSDStream(mqttClient, hubKey, payload, configuration, communication)
|
||||
case "request-hd-stream":
|
||||
go HandleRequestHDStream(mqttClient, hubKey, payload, configuration, communication)
|
||||
case "request-hls-stream":
|
||||
go HandleRequestHLSStream(mqttClient, hubKey, payload, configuration, communication)
|
||||
case "receive-hd-candidates":
|
||||
go HandleReceiveHDCandidates(mqttClient, hubKey, payload, configuration, communication)
|
||||
case "trigger-relay":
|
||||
@@ -559,6 +561,30 @@ func HandleRequestSDStream(mqttClient mqtt.Client, hubKey string, payload models
|
||||
}
|
||||
}
|
||||
|
||||
// HandleRequestHLSStream is the viewer keepalive for live HLS. Like the SD
|
||||
// stream it simply signals that a viewer is watching; the agent owns the live
|
||||
// HLS session, so a single non-zero timestamp on the channel keeps the segment
|
||||
// pipeline alive (see cloud.HandleLiveStreamHLS). Viewers republish this
|
||||
// periodically; when the keepalives stop, the agent tears the session down.
|
||||
func HandleRequestHLSStream(mqttClient mqtt.Client, hubKey string, payload models.Payload, configuration *models.Configuration, communication *models.Communication) {
|
||||
value := payload.Value
|
||||
jsonData, _ := json.Marshal(value)
|
||||
var requestHLSStreamPayload models.RequestHLSStreamPayload
|
||||
json.Unmarshal(jsonData, &requestHLSStreamPayload)
|
||||
|
||||
if requestHLSStreamPayload.Timestamp != 0 {
|
||||
if communication.CameraConnected {
|
||||
select {
|
||||
case communication.HandleLiveHLS <- time.Now().Unix():
|
||||
default:
|
||||
}
|
||||
log.Log.Info("routers.mqtt.main.HandleRequestHLSStream(): received request to livestream over HLS.")
|
||||
} else {
|
||||
log.Log.Info("routers.mqtt.main.HandleRequestHLSStream(): received request to livestream over HLS, but camera is not connected.")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func HandleRequestHDStream(mqttClient mqtt.Client, hubKey string, payload models.Payload, configuration *models.Configuration, communication *models.Communication) {
|
||||
value := payload.Value
|
||||
// Convert map[string]interface{} to RequestHDStreamPayload
|
||||
|
||||
367
machinery/src/video/livehls.go
Normal file
367
machinery/src/video/livehls.go
Normal file
@@ -0,0 +1,367 @@
|
||||
package video
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"fmt"
|
||||
|
||||
mp4ff "github.com/Eyevinn/mp4ff/mp4"
|
||||
"github.com/kerberos-io/agent/machinery/src/log"
|
||||
)
|
||||
|
||||
// LiveSegmenter turns a live stream of Annex B video samples into HLS-ready
|
||||
// fragmented-MP4 (CMAF) output: ONE init segment (ftyp+moov) followed by a
|
||||
// series of INDEPENDENT media segments (styp+moof+mdat), each beginning with a
|
||||
// keyframe and carrying its own tfdt. This is the building block for the live
|
||||
// HLS pipeline (agent -> hub-api -> vault -> hub-frontend) and is intentionally
|
||||
// kept separate from the recording muxer in mp4.go:
|
||||
//
|
||||
// - mp4.go writes ONE fragmented MP4 per recording (free-box placeholder up
|
||||
// front, back-filled on Close). That layout is great for archived files but
|
||||
// useless for live, where each segment must be shippable the instant it is
|
||||
// produced and must decode on its own after the init segment.
|
||||
// - LiveSegmenter emits discrete, self-contained segments via callbacks, so
|
||||
// the transport (single-POST to hub-api, drop-on-failure) never has to wait
|
||||
// for the recording to finish.
|
||||
//
|
||||
// Both producers use the SAME mp4ff fragment format, so live and archived video
|
||||
// share one toolchain on the player side (hls.js #EXT-X-MAP + byte-range parts).
|
||||
//
|
||||
// The spike scope is video-only H.264/H.265. Audio and multi-track interleaving
|
||||
// can be layered on later by adding tracks to the init segment and a second trun
|
||||
// to each fragment; nothing here precludes that.
|
||||
type LiveSegmenter struct {
|
||||
// codec is "H264"/"H265" (case handled in buildInit).
|
||||
codec string
|
||||
// timescale is the media timescale used in the init segment. The agent's
|
||||
// capture path feeds presentation timestamps in milliseconds, so a 1000-tick
|
||||
// timescale keeps sample durations exact with no rescaling.
|
||||
timescale uint32
|
||||
// targetSegmentMs is the minimum amount of media a segment accumulates before
|
||||
// the next keyframe is allowed to start a fresh segment. Keeping segments
|
||||
// keyframe-aligned is what makes each one independently decodable.
|
||||
targetSegmentMs uint64
|
||||
|
||||
spsNALUs [][]byte
|
||||
ppsNALUs [][]byte
|
||||
vpsNALUs [][]byte
|
||||
|
||||
// width/height are written into the visual sample entry. They are optional:
|
||||
// on a successful strict SPS parse mp4ff derives them, but the manual avcC
|
||||
// fallback (used for SPS that mp4ff cannot parse) needs them supplied.
|
||||
width uint16
|
||||
height uint16
|
||||
|
||||
videoTrackID uint32
|
||||
|
||||
initSegment *mp4ff.InitSegment
|
||||
initBytes []byte
|
||||
initEmitted bool
|
||||
|
||||
seg *mp4ff.MediaSegment
|
||||
frag *mp4ff.Fragment
|
||||
seqNr uint32
|
||||
|
||||
// started becomes true once the first segment has been opened.
|
||||
started bool
|
||||
// segStartPTS is the decode time (ms) of the first sample in the open
|
||||
// segment; elapsed media is measured against it to decide segment cuts.
|
||||
segStartPTS uint64
|
||||
// segDurationMs accumulates the committed sample durations of the open
|
||||
// segment so the playlist can advertise an accurate #EXTINF.
|
||||
segDurationMs uint64
|
||||
|
||||
// pending holds the most recently received sample. Its duration is only known
|
||||
// once the NEXT sample arrives (duration = nextPTS - thisPTS), mirroring the
|
||||
// pending-sample pattern used by the recording muxer.
|
||||
pending *mp4ff.FullSample
|
||||
// lastDurationMs is the previous committed duration, reused to close out the
|
||||
// final pending sample (and to bridge non-monotonic timestamps).
|
||||
lastDurationMs uint64
|
||||
|
||||
// OnInit is invoked exactly once with the encoded init segment bytes before
|
||||
// the first media segment is emitted. Optional.
|
||||
OnInit func(initBytes []byte) error
|
||||
// OnSegment is invoked once per completed media segment. Optional.
|
||||
OnSegment func(seg LiveSegment) error
|
||||
}
|
||||
|
||||
// LiveSegment is one independently-decodable CMAF media segment.
|
||||
type LiveSegment struct {
|
||||
// SequenceNumber is the monotonically increasing fragment sequence number
|
||||
// (also used as the moof sequence number and the seg-N.m4s index).
|
||||
SequenceNumber uint32
|
||||
// DurationMs is the summed sample duration of the segment, for #EXTINF.
|
||||
DurationMs uint64
|
||||
// Data is the complete styp+moof+mdat segment, ready to append after the init
|
||||
// segment and hand to hls.js / a vault object.
|
||||
Data []byte
|
||||
}
|
||||
|
||||
// Sample-entry flags matching the recording muxer so live and archived fragments
|
||||
// describe random access points identically.
|
||||
//
|
||||
// keyframe 0x02000000 = sampleDependsOn=2 (depends on nothing), sync sample
|
||||
// non-keyframe 0x01010000 = sampleDependsOn=1, sampleIsNonSyncSample=1
|
||||
const (
|
||||
liveSyncSampleFlags uint32 = 0x02000000
|
||||
liveNonSyncSampleFlags uint32 = 0x01010000
|
||||
// liveFallbackDurationMs is used when a duration cannot be derived (first
|
||||
// frame at Close, or non-monotonic timestamps) and no prior duration exists.
|
||||
// ~33 ms approximates 30 fps and is only ever a single-frame nicety.
|
||||
liveFallbackDurationMs uint64 = 33
|
||||
)
|
||||
|
||||
// NewLiveSegmenter creates a video-only live segmenter for the given codec.
|
||||
// spsNALUs/ppsNALUs (and vpsNALUs for H.265) may be raw NAL units or Annex B
|
||||
// blobs with start codes; both are normalized. targetSegmentMs is clamped to a
|
||||
// sane floor so a misconfiguration cannot produce one-frame segments.
|
||||
func NewLiveSegmenter(codec string, spsNALUs, ppsNALUs, vpsNALUs [][]byte, targetSegmentMs uint64) *LiveSegmenter {
|
||||
if targetSegmentMs < 500 {
|
||||
targetSegmentMs = 500
|
||||
}
|
||||
return &LiveSegmenter{
|
||||
codec: codec,
|
||||
timescale: 1000,
|
||||
targetSegmentMs: targetSegmentMs,
|
||||
spsNALUs: spsNALUs,
|
||||
ppsNALUs: ppsNALUs,
|
||||
vpsNALUs: vpsNALUs,
|
||||
}
|
||||
}
|
||||
|
||||
// SetDimensions records the encoded video width/height in pixels. They are
|
||||
// written into the avc1/hvc1 visual sample entry and are required for the manual
|
||||
// descriptor fallback path (SPS that mp4ff's strict parser rejects).
|
||||
func (ls *LiveSegmenter) SetDimensions(width, height uint16) {
|
||||
ls.width = width
|
||||
ls.height = height
|
||||
}
|
||||
|
||||
// InitSegment returns the encoded init segment bytes, building them on demand.
|
||||
// Useful for tests and for serving the #EXT-X-MAP target without waiting for the
|
||||
// first media segment.
|
||||
func (ls *LiveSegmenter) InitSegment() ([]byte, error) {
|
||||
if ls.initBytes == nil {
|
||||
if err := ls.buildInit(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return ls.initBytes, nil
|
||||
}
|
||||
|
||||
// buildInit constructs the ftyp+moov init segment from the parameter sets.
|
||||
func (ls *LiveSegmenter) buildInit() error {
|
||||
init := mp4ff.CreateEmptyInit()
|
||||
init.AddEmptyTrack(ls.timescale, "video", "und")
|
||||
trak := init.Moov.Traks[0]
|
||||
|
||||
switch ls.codec {
|
||||
case "H264", "h264", "AVC", "avc", "AVC1", "avc1":
|
||||
sps, pps := normalizeH264ParameterSets(ls.spsNALUs, ls.ppsNALUs)
|
||||
if len(sps) == 0 || len(pps) == 0 {
|
||||
return fmt.Errorf("livehls: missing H264 SPS/PPS (sps=%d pps=%d)", len(sps), len(pps))
|
||||
}
|
||||
// includePS=true stores SPS/PPS in the avcC so segments need not carry
|
||||
// in-band parameter sets - browsers read them from the init segment. Some
|
||||
// camera SPS variants trip mp4ff's strict parser (e.g. unusual VUI/SAR);
|
||||
// fall back to a manually built avcC just like the recording muxer does so
|
||||
// those cameras still produce a valid init segment.
|
||||
if err := trak.SetAVCDescriptor("avc1", sps, pps, true); err != nil {
|
||||
log.Log.Warning("livehls: SetAVCDescriptor failed, using manual avcC fallback: " + err.Error())
|
||||
if fbErr := addAVCDescriptorFallback(trak, sps, pps, ls.width, ls.height); fbErr != nil {
|
||||
return fmt.Errorf("livehls: AVC descriptor fallback: %w", fbErr)
|
||||
}
|
||||
}
|
||||
case "H265", "h265", "HEVC", "hevc", "HVC1", "hvc1":
|
||||
vps, sps, pps := normalizeH265ParameterSets(ls.vpsNALUs, ls.spsNALUs, ls.ppsNALUs)
|
||||
if len(vps) == 0 || len(sps) == 0 || len(pps) == 0 {
|
||||
return fmt.Errorf("livehls: missing H265 VPS/SPS/PPS (vps=%d sps=%d pps=%d)", len(vps), len(sps), len(pps))
|
||||
}
|
||||
if err := trak.SetHEVCDescriptor("hvc1", vps, sps, pps, [][]byte{}, true); err != nil {
|
||||
return fmt.Errorf("livehls: SetHEVCDescriptor: %w", err)
|
||||
}
|
||||
default:
|
||||
return fmt.Errorf("livehls: unsupported codec %q", ls.codec)
|
||||
}
|
||||
|
||||
// Record the encoded dimensions in the track header when known.
|
||||
if ls.width > 0 && ls.height > 0 {
|
||||
trak.Tkhd.Width = mp4ff.Fixed32(uint32(ls.width) << 16)
|
||||
trak.Tkhd.Height = mp4ff.Fixed32(uint32(ls.height) << 16)
|
||||
}
|
||||
// mdhd.Duration MUST be 0 for fragmented MP4 so players derive duration from
|
||||
// the fragments rather than a (here unknown) total.
|
||||
trak.Mdia.Mdhd.Duration = 0
|
||||
|
||||
ls.videoTrackID = trak.Tkhd.TrackID
|
||||
|
||||
var buf bytes.Buffer
|
||||
if err := init.Encode(&buf); err != nil {
|
||||
return fmt.Errorf("livehls: encode init: %w", err)
|
||||
}
|
||||
ls.initSegment = init
|
||||
ls.initBytes = buf.Bytes()
|
||||
return nil
|
||||
}
|
||||
|
||||
// WriteSample feeds one Annex B access unit with its decode timestamp (DTS) in
|
||||
// milliseconds. The first sample of a session MUST be a keyframe; a non-keyframe
|
||||
// first sample is dropped (it could not be decoded without a preceding IDR).
|
||||
//
|
||||
// compositionOffsetMs is the CTS offset (PTS-DTS, for B-frame reordering) in
|
||||
// timescale ticks; pass 0 for streams without B-frames.
|
||||
func (ls *LiveSegmenter) WriteSample(isKeyframe bool, annexB []byte, ptsMs uint64, compositionOffsetMs int32) error {
|
||||
// Lazily build + emit the init segment on the first accepted sample.
|
||||
if ls.initBytes == nil {
|
||||
if err := ls.buildInit(); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if !ls.initEmitted {
|
||||
ls.initEmitted = true
|
||||
if ls.OnInit != nil {
|
||||
if err := ls.OnInit(ls.initBytes); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A session must open on a random-access point; otherwise the first segment
|
||||
// would reference frames that never arrived.
|
||||
if !ls.started && !isKeyframe {
|
||||
log.Log.Debug("LiveSegmenter.WriteSample(): dropping leading non-keyframe before first IDR")
|
||||
return nil
|
||||
}
|
||||
|
||||
lengthPrefixed, err := annexBToLengthPrefixed(annexB)
|
||||
if err != nil {
|
||||
return fmt.Errorf("livehls: convert AnnexB: %w", err)
|
||||
}
|
||||
|
||||
// The previous sample's duration is the gap to this sample's PTS. Commit it
|
||||
// to the (still open) current fragment before we consider rolling segments,
|
||||
// because the pending sample always precedes this one in decode order.
|
||||
if ls.pending != nil {
|
||||
dur := ls.lastDurationMs
|
||||
if ptsMs > ls.pending.DecodeTime {
|
||||
dur = ptsMs - ls.pending.DecodeTime
|
||||
}
|
||||
if dur == 0 {
|
||||
dur = liveFallbackDurationMs
|
||||
}
|
||||
ls.lastDurationMs = dur
|
||||
ls.pending.Sample.Dur = uint32(dur)
|
||||
if err := ls.commitPending(); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// At every keyframe, decide whether enough media has accumulated to close the
|
||||
// open segment and start a new one. Cutting only on keyframes guarantees each
|
||||
// segment is independently decodable.
|
||||
if isKeyframe {
|
||||
shouldCut := !ls.started || (ptsMs-ls.segStartPTS) >= ls.targetSegmentMs
|
||||
if shouldCut {
|
||||
if ls.started {
|
||||
if err := ls.emitSegment(); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
ls.openSegment(ptsMs)
|
||||
}
|
||||
}
|
||||
|
||||
// Stage this sample; its duration is filled in when the next sample arrives
|
||||
// (or at Close()).
|
||||
flags := liveNonSyncSampleFlags
|
||||
if isKeyframe {
|
||||
flags = liveSyncSampleFlags
|
||||
}
|
||||
ls.pending = &mp4ff.FullSample{
|
||||
Sample: mp4ff.Sample{
|
||||
Flags: flags,
|
||||
Size: uint32(len(lengthPrefixed)),
|
||||
CompositionTimeOffset: compositionOffsetMs,
|
||||
},
|
||||
DecodeTime: ptsMs,
|
||||
Data: lengthPrefixed,
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// openSegment starts a fresh media segment (with CMAF styp) and an empty
|
||||
// single-track fragment whose moof sequence number is the segment index.
|
||||
func (ls *LiveSegmenter) openSegment(startPTS uint64) {
|
||||
ls.seqNr++
|
||||
ls.seg = mp4ff.NewMediaSegment() // includes a CMAF styp box by default
|
||||
frag, err := mp4ff.CreateFragment(ls.seqNr, ls.videoTrackID)
|
||||
if err != nil {
|
||||
log.Log.Error("LiveSegmenter.openSegment(): CreateFragment failed: " + err.Error())
|
||||
return
|
||||
}
|
||||
ls.seg.AddFragment(frag)
|
||||
ls.frag = frag
|
||||
ls.segStartPTS = startPTS
|
||||
ls.segDurationMs = 0
|
||||
ls.started = true
|
||||
}
|
||||
|
||||
// commitPending appends the staged sample to the open fragment. The first sample
|
||||
// of a fragment seeds the tfdt baseMediaDecodeTime from its absolute DecodeTime,
|
||||
// which is what makes the segment independently seekable/decodable.
|
||||
func (ls *LiveSegmenter) commitPending() error {
|
||||
if ls.pending == nil {
|
||||
return nil
|
||||
}
|
||||
if ls.frag == nil {
|
||||
// No open segment yet (e.g. pending set before the first keyframe cut). The
|
||||
// keyframe path always opens a segment before staging, so this only guards
|
||||
// against logic drift; drop rather than panic.
|
||||
ls.pending = nil
|
||||
return nil
|
||||
}
|
||||
if err := ls.frag.AddFullSampleToTrack(*ls.pending, ls.videoTrackID); err != nil {
|
||||
return fmt.Errorf("livehls: AddFullSampleToTrack: %w", err)
|
||||
}
|
||||
ls.segDurationMs += uint64(ls.pending.Sample.Dur)
|
||||
ls.pending = nil
|
||||
return nil
|
||||
}
|
||||
|
||||
// emitSegment encodes the open segment and hands it to OnSegment.
|
||||
func (ls *LiveSegmenter) emitSegment() error {
|
||||
if ls.seg == nil {
|
||||
return nil
|
||||
}
|
||||
var buf bytes.Buffer
|
||||
if err := ls.seg.Encode(&buf); err != nil {
|
||||
return fmt.Errorf("livehls: encode segment %d: %w", ls.seqNr, err)
|
||||
}
|
||||
out := LiveSegment{
|
||||
SequenceNumber: ls.seqNr,
|
||||
DurationMs: ls.segDurationMs,
|
||||
Data: buf.Bytes(),
|
||||
}
|
||||
ls.seg = nil
|
||||
ls.frag = nil
|
||||
if ls.OnSegment != nil {
|
||||
return ls.OnSegment(out)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Close flushes the final pending sample and emits the last open segment. Call
|
||||
// once when the live session ends so no trailing media is lost.
|
||||
func (ls *LiveSegmenter) Close() error {
|
||||
if ls.pending != nil {
|
||||
dur := ls.lastDurationMs
|
||||
if dur == 0 {
|
||||
dur = liveFallbackDurationMs
|
||||
}
|
||||
ls.pending.Sample.Dur = uint32(dur)
|
||||
if err := ls.commitPending(); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return ls.emitSegment()
|
||||
}
|
||||
371
machinery/src/video/livehls_test.go
Normal file
371
machinery/src/video/livehls_test.go
Normal file
@@ -0,0 +1,371 @@
|
||||
package video
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"fmt"
|
||||
"math"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
mp4ff "github.com/Eyevinn/mp4ff/mp4"
|
||||
)
|
||||
|
||||
// Known-good minimal H.264 baseline parameter sets (640x480), reused from the
|
||||
// recording-muxer tests so the live segmenter is exercised against the exact
|
||||
// SPS/PPS mp4ff is already known to parse into an avcC descriptor.
|
||||
var (
|
||||
liveTestSPS = []byte{0x67, 0x42, 0xc0, 0x1e, 0xd9, 0x00, 0xa0, 0x47, 0xfe, 0xc8}
|
||||
liveTestPPS = []byte{0x68, 0xce, 0x38, 0x80}
|
||||
)
|
||||
|
||||
// makeAnnexBFrame builds a single-NALU Annex B access unit: a 4-byte start code,
|
||||
// the NAL header (IDR=0x65 for keyframes, non-IDR=0x01 otherwise) and padding.
|
||||
func makeAnnexBFrame(isKey bool) []byte {
|
||||
nalType := byte(0x01)
|
||||
if isKey {
|
||||
nalType = 0x65
|
||||
}
|
||||
frame := []byte{0x00, 0x00, 0x00, 0x01, nalType}
|
||||
for i := 0; i < 100; i++ {
|
||||
frame = append(frame, byte(i))
|
||||
}
|
||||
return frame
|
||||
}
|
||||
|
||||
// isSyncSample reports whether a parsed sample is a random-access point
|
||||
// (sample_depends_on == 2 => "depends on nothing" => IDR/sync).
|
||||
func isSyncSample(s mp4ff.Sample) bool {
|
||||
return (s.Flags>>24)&0x03 == 0x02
|
||||
}
|
||||
|
||||
// TestLiveSegmenterProducesIndependentCMAFSegments feeds a synthetic H.264
|
||||
// stream (25 fps, 1s GOPs) through the live segmenter and asserts that:
|
||||
// - exactly one init segment (ftyp+moov, single avc1 video track) is produced;
|
||||
// - segments are cut on keyframe boundaries honoring the target duration;
|
||||
// - every media segment carries a CMAF styp + exactly one moof+mdat fragment;
|
||||
// - each segment begins with a sync sample and its tfdt equals the absolute
|
||||
// decode time of that first sample (the property that makes it independently
|
||||
// decodable after the init segment);
|
||||
// - sample counts and durations are preserved end to end.
|
||||
func TestLiveSegmenterProducesIndependentCMAFSegments(t *testing.T) {
|
||||
const (
|
||||
frameDurMs = uint64(40) // 25 fps
|
||||
gopFrames = 25 // keyframe every 1000 ms
|
||||
numGOPs = 6
|
||||
numFrames = gopFrames * numGOPs // 150 frames, 6000 ms
|
||||
targetMs = uint64(2000) // 2s segments => 2 GOPs each
|
||||
)
|
||||
|
||||
seg := NewLiveSegmenter("H264", [][]byte{liveTestSPS}, [][]byte{liveTestPPS}, nil, targetMs)
|
||||
seg.SetDimensions(640, 480)
|
||||
|
||||
var initBytes []byte
|
||||
var initCalls int
|
||||
var segments []LiveSegment
|
||||
seg.OnInit = func(b []byte) error {
|
||||
initCalls++
|
||||
initBytes = append([]byte(nil), b...)
|
||||
return nil
|
||||
}
|
||||
seg.OnSegment = func(s LiveSegment) error {
|
||||
segments = append(segments, s)
|
||||
return nil
|
||||
}
|
||||
|
||||
for i := 0; i < numFrames; i++ {
|
||||
isKey := i%gopFrames == 0
|
||||
pts := uint64(i) * frameDurMs
|
||||
if err := seg.WriteSample(isKey, makeAnnexBFrame(isKey), pts, 0); err != nil {
|
||||
t.Fatalf("WriteSample(frame=%d): %v", i, err)
|
||||
}
|
||||
}
|
||||
if err := seg.Close(); err != nil {
|
||||
t.Fatalf("Close: %v", err)
|
||||
}
|
||||
|
||||
// --- Init segment: emitted exactly once, well-formed, single video track. ---
|
||||
if initCalls != 1 {
|
||||
t.Fatalf("OnInit called %d times, want 1", initCalls)
|
||||
}
|
||||
if len(initBytes) == 0 {
|
||||
t.Fatal("init segment is empty")
|
||||
}
|
||||
parsedInit, err := mp4ff.DecodeFile(bytes.NewReader(initBytes))
|
||||
if err != nil {
|
||||
t.Fatalf("decode init: %v", err)
|
||||
}
|
||||
if parsedInit.Init == nil || parsedInit.Init.Ftyp == nil || parsedInit.Init.Moov == nil {
|
||||
t.Fatal("init segment missing ftyp/moov")
|
||||
}
|
||||
if got := len(parsedInit.Init.Moov.Traks); got != 1 {
|
||||
t.Fatalf("init moov has %d traks, want 1", got)
|
||||
}
|
||||
|
||||
// --- Segment cut cadence: 6 GOPs at 2s target => 3 segments of 2 GOPs each. ---
|
||||
const wantSegments = 3
|
||||
if len(segments) != wantSegments {
|
||||
t.Fatalf("got %d media segments, want %d", len(segments), wantSegments)
|
||||
}
|
||||
for i, s := range segments {
|
||||
if want := uint32(i + 1); s.SequenceNumber != want {
|
||||
t.Errorf("segment %d: SequenceNumber=%d, want %d", i, s.SequenceNumber, want)
|
||||
}
|
||||
if s.DurationMs != targetMs {
|
||||
t.Errorf("segment %d: DurationMs=%d, want %d", i, s.DurationMs, targetMs)
|
||||
}
|
||||
}
|
||||
|
||||
// --- Each segment must decode INDEPENDENTLY after the init segment. ---
|
||||
// Parsing init+oneSegment in isolation mirrors exactly what hls.js does with
|
||||
// an #EXT-X-MAP init and a single media part.
|
||||
var totalSamples, totalSync int
|
||||
wantTFDT := []uint64{0, 2000, 4000}
|
||||
for i, s := range segments {
|
||||
standalone := append(append([]byte(nil), initBytes...), s.Data...)
|
||||
parsed, err := mp4ff.DecodeFile(bytes.NewReader(standalone))
|
||||
if err != nil {
|
||||
t.Fatalf("segment %d: decode init+segment: %v", i, err)
|
||||
}
|
||||
if len(parsed.Segments) != 1 {
|
||||
t.Fatalf("segment %d: parsed %d media segments, want 1", i, len(parsed.Segments))
|
||||
}
|
||||
mseg := parsed.Segments[0]
|
||||
if mseg.Styp == nil {
|
||||
t.Errorf("segment %d: missing CMAF styp box", i)
|
||||
}
|
||||
if len(mseg.Fragments) != 1 {
|
||||
t.Fatalf("segment %d: %d fragments, want 1", i, len(mseg.Fragments))
|
||||
}
|
||||
fr := mseg.Fragments[0]
|
||||
if got := fr.Moof.Mfhd.SequenceNumber; got != s.SequenceNumber {
|
||||
t.Errorf("segment %d: moof sequence=%d, want %d", i, got, s.SequenceNumber)
|
||||
}
|
||||
traf := fr.Moof.Traf
|
||||
if traf.Tfhd.TrackID != 1 {
|
||||
t.Errorf("segment %d: track id=%d, want 1", i, traf.Tfhd.TrackID)
|
||||
}
|
||||
if got := traf.Tfdt.BaseMediaDecodeTime(); got != wantTFDT[i] {
|
||||
t.Errorf("segment %d: tfdt baseMediaDecodeTime=%d, want %d", i, got, wantTFDT[i])
|
||||
}
|
||||
|
||||
var samples []mp4ff.Sample
|
||||
for _, trun := range traf.Truns {
|
||||
samples = append(samples, trun.Samples...)
|
||||
}
|
||||
if len(samples) == 0 {
|
||||
t.Fatalf("segment %d: no samples", i)
|
||||
}
|
||||
if !isSyncSample(samples[0]) {
|
||||
t.Errorf("segment %d: first sample is not a keyframe/sync sample", i)
|
||||
}
|
||||
var segDur uint64
|
||||
for j, smp := range samples {
|
||||
totalSamples++
|
||||
if isSyncSample(smp) {
|
||||
totalSync++
|
||||
}
|
||||
segDur += uint64(smp.Dur)
|
||||
if smp.Size == 0 {
|
||||
t.Errorf("segment %d sample %d: zero size", i, j)
|
||||
}
|
||||
}
|
||||
if segDur != s.DurationMs {
|
||||
t.Errorf("segment %d: summed sample dur=%d, reported DurationMs=%d", i, segDur, s.DurationMs)
|
||||
}
|
||||
}
|
||||
|
||||
if totalSamples != numFrames {
|
||||
t.Errorf("total samples across segments=%d, want %d", totalSamples, numFrames)
|
||||
}
|
||||
if totalSync != numGOPs {
|
||||
t.Errorf("total sync samples=%d, want %d (one per GOP)", totalSync, numGOPs)
|
||||
}
|
||||
}
|
||||
|
||||
// TestLiveSegmenterDropsLeadingNonKeyframe verifies a session cannot open on a
|
||||
// non-IDR frame (which would reference frames that never arrived); such leading
|
||||
// samples are dropped until the first keyframe.
|
||||
func TestLiveSegmenterDropsLeadingNonKeyframe(t *testing.T) {
|
||||
seg := NewLiveSegmenter("H264", [][]byte{liveTestSPS}, [][]byte{liveTestPPS}, nil, 1000)
|
||||
seg.SetDimensions(640, 480)
|
||||
var segments []LiveSegment
|
||||
seg.OnSegment = func(s LiveSegment) error { segments = append(segments, s); return nil }
|
||||
|
||||
// Two P-frames before any IDR must be ignored.
|
||||
if err := seg.WriteSample(false, makeAnnexBFrame(false), 0, 0); err != nil {
|
||||
t.Fatalf("WriteSample(p0): %v", err)
|
||||
}
|
||||
if err := seg.WriteSample(false, makeAnnexBFrame(false), 40, 0); err != nil {
|
||||
t.Fatalf("WriteSample(p1): %v", err)
|
||||
}
|
||||
// First IDR opens the session at decode time 0.
|
||||
for i := 0; i < 25; i++ {
|
||||
isKey := i == 0
|
||||
if err := seg.WriteSample(isKey, makeAnnexBFrame(isKey), uint64(i)*40, 0); err != nil {
|
||||
t.Fatalf("WriteSample(%d): %v", i, err)
|
||||
}
|
||||
}
|
||||
if err := seg.Close(); err != nil {
|
||||
t.Fatalf("Close: %v", err)
|
||||
}
|
||||
|
||||
if len(segments) == 0 {
|
||||
t.Fatal("expected at least one segment after the first IDR")
|
||||
}
|
||||
initBytes, err := seg.InitSegment()
|
||||
if err != nil {
|
||||
t.Fatalf("InitSegment: %v", err)
|
||||
}
|
||||
standalone := append(append([]byte(nil), initBytes...), segments[0].Data...)
|
||||
parsed, err := mp4ff.DecodeFile(bytes.NewReader(standalone))
|
||||
if err != nil {
|
||||
t.Fatalf("decode: %v", err)
|
||||
}
|
||||
traf := parsed.Segments[0].Fragments[0].Moof.Traf
|
||||
if got := traf.Tfdt.BaseMediaDecodeTime(); got != 0 {
|
||||
t.Errorf("first segment tfdt=%d, want 0 (session opens on the IDR)", got)
|
||||
}
|
||||
var first mp4ff.Sample
|
||||
for _, trun := range traf.Truns {
|
||||
if len(trun.Samples) > 0 {
|
||||
first = trun.Samples[0]
|
||||
break
|
||||
}
|
||||
}
|
||||
if !isSyncSample(first) {
|
||||
t.Error("first committed sample must be the IDR, not a dropped P-frame")
|
||||
}
|
||||
}
|
||||
|
||||
// renderLiveMediaPlaylist renders a live (no #EXT-X-ENDLIST) fMP4 HLS media
|
||||
// playlist for the given segments. This mirrors the shape hub-api will serve for
|
||||
// live streams: an #EXT-X-MAP init segment followed by one #EXTINF per CMAF part.
|
||||
// In production hub-api emits a sliding WINDOW of the most recent segments and
|
||||
// advances #EXT-X-MEDIA-SEQUENCE; here we list the whole synthetic capture for a
|
||||
// self-contained, inspectable bundle.
|
||||
func renderLiveMediaPlaylist(initURI string, segs []LiveSegment, mediaSequence uint32) string {
|
||||
var maxDurMs uint64
|
||||
for _, s := range segs {
|
||||
if s.DurationMs > maxDurMs {
|
||||
maxDurMs = s.DurationMs
|
||||
}
|
||||
}
|
||||
target := uint64(math.Ceil(float64(maxDurMs) / 1000.0))
|
||||
if target == 0 {
|
||||
target = 1
|
||||
}
|
||||
|
||||
var b strings.Builder
|
||||
b.WriteString("#EXTM3U\n")
|
||||
b.WriteString("#EXT-X-VERSION:7\n")
|
||||
fmt.Fprintf(&b, "#EXT-X-TARGETDURATION:%d\n", target)
|
||||
fmt.Fprintf(&b, "#EXT-X-MEDIA-SEQUENCE:%d\n", mediaSequence)
|
||||
b.WriteString("#EXT-X-INDEPENDENT-SEGMENTS\n")
|
||||
fmt.Fprintf(&b, "#EXT-X-MAP:URI=%q\n", initURI)
|
||||
for _, s := range segs {
|
||||
fmt.Fprintf(&b, "#EXTINF:%.3f,\n", float64(s.DurationMs)/1000.0)
|
||||
fmt.Fprintf(&b, "seg-%d.m4s\n", s.SequenceNumber)
|
||||
}
|
||||
// NOTE: deliberately no #EXT-X-ENDLIST - its absence is what marks the
|
||||
// playlist as live so hls.js keeps polling for new segments.
|
||||
return b.String()
|
||||
}
|
||||
|
||||
// TestLiveSegmenterWritesHLSBundle runs the segmenter over a synthetic stream and
|
||||
// writes a complete on-disk fMP4 HLS bundle (init.mp4 + seg-N.m4s + a live
|
||||
// stream.m3u8). It validates the playlist shape and that every referenced file
|
||||
// exists, then logs the output directory so the structure can be eyeballed.
|
||||
//
|
||||
// Set LIVEHLS_OUT=/some/dir to keep the bundle for manual inspection (e.g. serve
|
||||
// it and point hls.js at stream.m3u8); otherwise a temp dir is used and removed.
|
||||
//
|
||||
// The frames here are synthetic (valid fMP4 boxing, non-decodable payloads), so
|
||||
// this validates CONTAINER/playlist structure, not pixel decode - the round-trip
|
||||
// assertions in TestLiveSegmenterProducesIndependentCMAFSegments cover decodable
|
||||
// box layout.
|
||||
func TestLiveSegmenterWritesHLSBundle(t *testing.T) {
|
||||
const (
|
||||
frameDurMs = uint64(40)
|
||||
gopFrames = 25
|
||||
numGOPs = 6
|
||||
numFrames = gopFrames * numGOPs
|
||||
targetMs = uint64(2000)
|
||||
)
|
||||
|
||||
outDir := os.Getenv("LIVEHLS_OUT")
|
||||
if outDir == "" {
|
||||
outDir = t.TempDir()
|
||||
} else {
|
||||
if err := os.MkdirAll(outDir, 0o755); err != nil {
|
||||
t.Fatalf("mkdir %s: %v", outDir, err)
|
||||
}
|
||||
}
|
||||
|
||||
seg := NewLiveSegmenter("H264", [][]byte{liveTestSPS}, [][]byte{liveTestPPS}, nil, targetMs)
|
||||
seg.SetDimensions(640, 480)
|
||||
|
||||
var segments []LiveSegment
|
||||
seg.OnInit = func(b []byte) error {
|
||||
return os.WriteFile(filepath.Join(outDir, "init.mp4"), b, 0o644)
|
||||
}
|
||||
seg.OnSegment = func(s LiveSegment) error {
|
||||
segments = append(segments, s)
|
||||
name := fmt.Sprintf("seg-%d.m4s", s.SequenceNumber)
|
||||
return os.WriteFile(filepath.Join(outDir, name), s.Data, 0o644)
|
||||
}
|
||||
|
||||
for i := 0; i < numFrames; i++ {
|
||||
isKey := i%gopFrames == 0
|
||||
if err := seg.WriteSample(isKey, makeAnnexBFrame(isKey), uint64(i)*frameDurMs, 0); err != nil {
|
||||
t.Fatalf("WriteSample(%d): %v", i, err)
|
||||
}
|
||||
}
|
||||
if err := seg.Close(); err != nil {
|
||||
t.Fatalf("Close: %v", err)
|
||||
}
|
||||
if len(segments) == 0 {
|
||||
t.Fatal("no segments produced")
|
||||
}
|
||||
|
||||
playlist := renderLiveMediaPlaylist("init.mp4", segments, segments[0].SequenceNumber)
|
||||
if err := os.WriteFile(filepath.Join(outDir, "stream.m3u8"), []byte(playlist), 0o644); err != nil {
|
||||
t.Fatalf("write playlist: %v", err)
|
||||
}
|
||||
|
||||
// --- Validate the live playlist shape. ---
|
||||
mustContain := []string{
|
||||
"#EXTM3U",
|
||||
"#EXT-X-VERSION:7",
|
||||
"#EXT-X-TARGETDURATION:2",
|
||||
"#EXT-X-MEDIA-SEQUENCE:1",
|
||||
`#EXT-X-MAP:URI="init.mp4"`,
|
||||
"#EXT-X-INDEPENDENT-SEGMENTS",
|
||||
}
|
||||
for _, tag := range mustContain {
|
||||
if !strings.Contains(playlist, tag) {
|
||||
t.Errorf("playlist missing %q\n---\n%s", tag, playlist)
|
||||
}
|
||||
}
|
||||
if strings.Contains(playlist, "#EXT-X-ENDLIST") {
|
||||
t.Error("live playlist must NOT contain #EXT-X-ENDLIST")
|
||||
}
|
||||
if got, want := strings.Count(playlist, "#EXTINF:"), len(segments); got != want {
|
||||
t.Errorf("playlist has %d #EXTINF entries, want %d", got, want)
|
||||
}
|
||||
|
||||
// --- Every referenced file must exist on disk. ---
|
||||
if _, err := os.Stat(filepath.Join(outDir, "init.mp4")); err != nil {
|
||||
t.Errorf("init.mp4 missing: %v", err)
|
||||
}
|
||||
for _, s := range segments {
|
||||
name := fmt.Sprintf("seg-%d.m4s", s.SequenceNumber)
|
||||
if _, err := os.Stat(filepath.Join(outDir, name)); err != nil {
|
||||
t.Errorf("%s missing: %v", name, err)
|
||||
}
|
||||
}
|
||||
|
||||
t.Logf("wrote HLS bundle to %s (%d segments)\n%s", outDir, len(segments), playlist)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user