diff --git a/CMakeLists.txt b/CMakeLists.txt index b2f3957..cc521f3 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -17,6 +17,18 @@ set(CMAKE_CXX_EXTENSIONS OFF) option(OPENMOQ_BUILD_TESTS "Build OpenMOQ publisher tests" ON) option(OPENMOQ_ENABLE_PICOQUIC "Enable picoquic transport integration when picoquic is available" ON) option(OPENMOQ_RUN_PICOQUIC_SMOKE_TESTS "Build and run picoquic loopback smoke tests" OFF) +option(OPENMOQ_ENABLE_SRT "Enable libsrt ingest support when libsrt is available" ON) + +set(OPENMOQ_HAS_SRT OFF) +if(OPENMOQ_ENABLE_SRT) + find_path(OPENMOQ_SRT_INCLUDE_DIR NAMES srt/srt.h) + find_library(OPENMOQ_SRT_LIBRARY NAMES srt) + if(OPENMOQ_SRT_INCLUDE_DIR AND OPENMOQ_SRT_LIBRARY) + set(OPENMOQ_HAS_SRT ON) + else() + message(STATUS "libsrt not found; building without SRT ingest runtime") + endif() +endif() set(_OPENMOQ_THIRDPARTY_ROOT "${CMAKE_CURRENT_SOURCE_DIR}/third_party") if(NOT EXISTS "${_OPENMOQ_THIRDPARTY_ROOT}") @@ -134,6 +146,8 @@ add_library(openmoq_publisher_lib src/cli_options.cpp src/cmaf_segmenter.cpp src/cmsf_packager.cpp + src/live_srt_ingest.cpp + src/live_srt_config.cpp src/moq_draft.cpp src/mp4_box.cpp src/publisher_api.cpp @@ -169,6 +183,12 @@ if(OPENMOQ_HAS_PICOQUIC) endif() endif() +if(OPENMOQ_HAS_SRT) + target_compile_definitions(openmoq_publisher_lib PRIVATE OPENMOQ_HAS_SRT=1) + target_include_directories(openmoq_publisher_lib PRIVATE ${OPENMOQ_SRT_INCLUDE_DIR}) + target_link_libraries(openmoq_publisher_lib PRIVATE ${OPENMOQ_SRT_LIBRARY}) +endif() + if(CMAKE_CXX_COMPILER_ID MATCHES "Clang|GNU") target_compile_options(openmoq_publisher_lib PRIVATE -Wall -Wextra -Wpedantic) elseif(MSVC) @@ -198,6 +218,12 @@ if(OPENMOQ_BUILD_TESTS) target_link_libraries(openmoq-publisher-cli-tests PRIVATE openmoq_publisher_lib) add_test(NAME openmoq-publisher-cli-tests COMMAND openmoq-publisher-cli-tests) + add_executable(openmoq-publisher-live-srt-config-tests + tests/live_srt_config_test.cpp + ) + target_link_libraries(openmoq-publisher-live-srt-config-tests PRIVATE openmoq_publisher_lib) + add_test(NAME openmoq-publisher-live-srt-config-tests COMMAND openmoq-publisher-live-srt-config-tests) + add_executable(openmoq-publisher-transport-tests tests/moqt_session_test.cpp ) diff --git a/README.md b/README.md index e035a67..d353f5a 100644 --- a/README.md +++ b/README.md @@ -51,6 +51,91 @@ OPENMOQ_PICOQUIC_TRACE=1 ./build/openmoq-publisher \ --paced ``` +Live ingest examples (choose one path, not both): + +1. SRT ingest path (`--live-source srt`) + +Create an SRT config file (example: `/tmp/srt_callers.json`): + +```json +{ + "srt_callers": [ + { + "id": "cam1", + "srt": { + "mode": "caller", + "host": "127.0.0.1", + "port": 9000, + "latency_ms": 120 + }, + "mpegts": { + "auto_detect_program": true, + "program_number": null, + "video_pid": null, + "audio_pid": null + }, + "cmaf": { + "fragment_on_keyframe": true, + "empty_moov": true, + "default_base_moof": true, + "separate_moof_per_track": true, + "target_fragment_duration_ms": 1000 + } + } + ] +} +``` + +Start the publisher (SRT receiver + MoQ publisher): + +```bash +./build/openmoq-publisher \ + --live-source srt \ + --srt-config /tmp/srt_callers.json \ + --endpoint 127.0.0.1:4443 \ + --transport raw \ + --namespace live \ + --draft 16 \ + --timeout 120 \ + --forward 0 +``` + +Feed MPEG-TS over SRT from ffmpeg: + +```bash +ffmpeg -hide_banner -stream_loop -1 -re \ + -i /home/ubuntu/bbb_sunflower_1080p_30fps_normal.mp4 \ + -filter_complex "[0:v]drawtext=fontcolor=white:fontsize=36:box=1:boxcolor=black@0.45:boxborderw=8:x=w-tw-20:y=20:text='%{localtime\\:%Y-%m-%d %T}\\:%{eif\\:mod(t*1000\\,1000)\\:d\\:3}'[vclock]" \ + -map "[vclock]" -map 0:a:0 \ + -c:v libx265 -preset veryfast -r 30 -g 60 -keyint_min 60 -bf 0 \ + -x265-params "keyint=60:min-keyint=60:scenecut=0:open-gop=0:repeat-headers=1" \ + -c:a aac -b:a 160k -ar 48000 -ac 2 \ + -f mpegts "srt://0.0.0.0:9000?mode=listener&pkt_size=1316" +``` + +2. stdin fragmented-MP4 path (`--live-source stdin`) + +```bash +ffmpeg -i /home/ubuntu/bbb_sunflower_1080p_30fps_normal.mp4 \ + -map 0:v:0 -map 0:a:0 \ + -map_metadata -1 \ + -sn -dn \ + -c:v libx264 -preset medium -r 30 -g 60 -keyint_min 60 -sc_threshold 0 -bf 0 \ + -c:a aac -b:a 160k -ar 48000 -ac 2 \ + -movflags +frag_keyframe+empty_moov+default_base_moof+separate_moof \ + -f mp4 - | ./build/openmoq-publisher \ + --live-source stdin \ + --input - \ + --endpoint 127.0.0.1:4443 \ + --transport raw \ + --namespace live \ + --draft 16 \ + --timeout 120 \ + --forward 0 +``` + +`--live-source both` is intentionally not supported. + On Windows, replace `./build/openmoq-publisher` with `build\Release\openmoq-publisher.exe` or the matching build configuration path. ## Documentation diff --git a/docs/srt-ingest-technical-note.md b/docs/srt-ingest-technical-note.md new file mode 100644 index 0000000..682272d --- /dev/null +++ b/docs/srt-ingest-technical-note.md @@ -0,0 +1,424 @@ +# SRT Live Ingest → MoQ Publishing: Technical Note + +## Overview +This document describes how `moqxr` receives live MPEG-TS over SRT, demuxes it into elementary stream samples, and publishes the media as MoQ objects. It also compares this to the stdin fragmented-MP4 path. + +--- + +## High-Level Pipeline + +``` +SRT socket (1316-byte datagrams) +└── MPEG-TS byte stream (buffered) + └── 188-byte TS packets (sync byte 0x47) + ├── PAT → discovers PMT PID + ├── PMT → discovers video PID + audio PID + stream types + └── PES assembly (per-PID) + └── EsSample (one complete access unit) + ├── Annex-B → MP4 length-prefixed NALUs (video) + │ or ADTS → raw AAC frames (audio) + └── moof + mdat (one sample per fragment) + └── MoQ Object (group_id / object_id) +``` + +--- + +## Step-by-Step Flow + +### 1. SRT Reception +**File:** `live_srt_ingest.cpp`, worker thread (line ~1340) + +``` +std::array recv_buf{}; +const int received = srt_recv(sock, recv_buf.data(), recv_buf.size()); +``` + +- Each `srt_recv()` returns up to **1316 bytes** — this is the standard SRT payload size (7 × 188-byte TS packets). +- The raw bytes are passed directly to the demuxer: `demuxer.feed(recv_buf.data(), received, sample_sink)`. + +--- + +### 2. Buffering and TS Packet Extraction +**Class:** `TsPesDemuxer::feed()` (line ~898) + +``` +buffer_.insert(buffer_.end(), data, data + size); + +while (buffer_.size() >= 188) { + if (buffer_[0] != 0x47) { + // Sync recovery: skip bytes until next 0x47 + auto sync_it = std::find(buffer_.begin() + 1, buffer_.end(), 0x47); + buffer_.erase(buffer_.begin(), sync_it); + continue; + } + // Extract one 188-byte TS packet + std::copy_n(buffer_.begin(), 188, packet.begin()); + buffer_.erase(buffer_.begin(), buffer_.begin() + 188); + parse_packet(packet, sample_sink); +} +``` +Incoming 1316-byte SRT payloads are appended to a persistent buffer. The buffer is then consumed 188 bytes at a time, each time verifying the sync byte `0x47`. If sync is lost, bytes are discarded until the next `0x47`. + +--- + +### 3. PAT Parsing +**Method:** `TsPesDemuxer::parse_pat()` (line ~946) + +- PID 0x0000 is always the PAT (Program Association Table). +- PAT entries map `program_number` → `PMT PID`. +- The code picks the first program (or the one matching a configured `program_number`). +- Result: `pmt_pid_` is set. + +--- + +### 4. PMT Parsing and PID Discovery +**Method:** `TsPesDemuxer::parse_pmt()` (line ~972) + +- When a TS packet arrives on the discovered `pmt_pid_`, the PMT (Program Map Table) is parsed. +- Each elementary stream descriptor has:`stream_type` (0x1B = H.264, 0x24 = HEVC, 0x0F = AAC-ADTS, etc.) +- `elementary_PID` + +- First video-type PID found → `video_pid_`, `video_stream_type_` +- First audio-type PID found → `audio_pid_`, `audio_stream_type_` +- After PMT: `pmt_parsed_ = true` + +**Stream type mapping:** + +Stream Type | Codec +--- | --- +0x02 | MPEG-2 Video +0x1B | H.264/AVC +0x24 | H.265/HEVC +0x0F | AAC (ADTS) +0x11 | AAC (LATM) +0x03/0x04 | MPEG Audio + +--- + +### 5. PES Packet Reconstruction +**Method:** `TsPesDemuxer::parse_pes()` (line ~1010) + +TS packets carrying a known video/audio PID are assembled into PES (Packetized Elementary Stream) buffers: + +1. When `payload_unit_start` is set → flush the previous PES buffer, start a new one +2. Parse the PES header:Bytes `[0..2]` = start code `00 00 01` +3. Byte 7 = flags (bit 7 = PTS present) +4. Byte 8 = PES header data length +5. If PTS present: extract 33-bit PTS from 5 bytes at offset 9 +6. Remaining payload bytes are appended to the PES data buffer +7. Subsequent TS packets (same PID, no `payload_unit_start`) append their payload to the same buffer + +**PTS extraction (33-bit timestamp at 90kHz clock):** + +``` +pes.pts90k = ((payload[9] >> 1) & 0x07) << 30 + | payload[10] << 22 + | ((payload[11] >> 1) & 0x7F) << 15 + | payload[12] << 7 + | ((payload[13] >> 1) & 0x7F); +``` + +--- + +### 6. EsSample — The Elementary Stream Access Unit +**Struct:** `EsSample` (line ~405) + +``` +struct EsSample { + bool is_video = false; + std::uint64_t pts90k = 0; // Presentation timestamp in 90kHz ticks + std::uint8_t stream_type = 0; // MPEG-TS stream_type from PMT + std::vector payload; // Raw ES data (length-prefixed NALUs for video, raw AAC for audio) + bool keyframe = false; + // For audio from ADTS: first 9 bytes of the original ADTS header (for codec discovery). + std::array adts_header{}; + std::uint8_t adts_header_len = 0; +}; +``` +When a PES buffer is flushed (`flush_pes`), one or more `EsSample`s are produced: + +- **Video:** one `EsSample` per PES (one coded picture in Annex-B format) +- **Audio:** the PES may contain multiple ADTS frames; `flush_pes` splits them into **individual `EsSample`s** with interpolated PTS (see §8) +- `keyframe` is determined by scanning for: + - H.264: NAL type 5 (IDR) + - HEVC: NAL types 16–23 (IRAP) + +--- + +### 7. Annex-B → MP4 Length-Prefixed (Video) +**Function:** `annexb_to_avcc()` (line ~355) + +MPEG-TS delivers H.264/HEVC in **Annex-B** format (start-code delimited): + +``` +00 00 00 01 [NAL] 00 00 00 01 [NAL] ... +``` +MP4/CMAF requires **length-prefixed** format: + +``` +[4-byte big-endian length] [NAL] [4-byte length] [NAL] ... +``` +The function: + +1. Scans for start codes (3-byte `00 00 01` or 4-byte `00 00 00 01`) +2. Measures the NAL unit length (bytes until next start code or end) +3. Writes `[BE32 length][NAL bytes]` for each unit + +--- + +### 8. AAC ADTS Handling +**Inline in:** `flush_pes()` (line ~1050) + +AAC in MPEG-TS uses ADTS (Audio Data Transport Stream) framing: + +``` +[ADTS header (7 or 9 bytes)] [AAC frame] [ADTS header] [AAC frame] ... +``` +MP4/CMAF stores **raw AAC frames** without ADTS headers. The ADTS splitting logic in `flush_pes()`: + +1. Validates sync word `0xFFF` (`0xFF` + upper nibble `0xF0`) at each frame start +2. Parses `protection_absent` flag → header is 7 bytes (no CRC) or 9 bytes (with CRC) +3. Reads `frame_length` field from ADTS header +4. Strips headers, emits each raw AAC frame as a **separate `EsSample`** +5. Interpolates PTS for each frame using rational arithmetic: `base_pts + (frame_index × 1024 × 90000 + rate/2) / rate` +6. Preserves the first ADTS header in `EsSample::adts_header` for codec discovery + +--- + +### 9. Building avcC / hvcC / esds Init Metadata +During codec discovery (before streaming begins), the code extracts decoder configuration from the first keyframe/ADTS frame: + +#### H.264 → avcC box +**Function:** `build_avcc_box()` (line ~590) + +1. `extract_h264_sps_pps()` scans the first keyframe's Annex-B data for NAL type 7 (SPS) and type 8 (PPS) +2. Builds an `AVCDecoderConfigurationRecord`:configurationVersion = 1 +3. profile/level from SPS bytes [1..3] +4. lengthSizeMinusOne = 3 (4-byte NAL lengths) +5. SPS array, PPS array + +#### HEVC → hvcC box +**Function:** `build_hvcc_box()` (line ~632) + +1. `extract_hevc_param_sets()` scans for NAL types 32 (VPS), 33 (SPS), 34 (PPS) +2. Builds an `HEVCDecoderConfigurationRecord`:Profile/tier/level parsed from SPS +3. 3 arrays: VPS, SPS, PPS + +#### AAC → esds box +**Function:** `build_esds_box()` (line ~728) + +1. Parses ADTS header: `profile`, `freq_index`, `channel_config` +2. Builds `AudioSpecificConfig` (2 bytes) +3. Wraps in nested MPEG-4 descriptors:ES_Descriptor (tag 0x03)DecoderConfigDescriptor (tag 0x04, objectType=0x40 = AAC)DecoderSpecificInfo (tag 0x05) = AudioSpecificConfig +4. SLConfigDescriptor (tag 0x06) + +--- + +### 10. Synthetic fMP4 Init Segment +**Function:** `build_init_segment_from_tracks()` (line ~320) + +After codec discovery completes (all `codec_private` bytes extracted), a synthetic fMP4 initialization segment is built: + +``` +ftyp (isom, iso6, mp41) +moov +├── mvhd (movie header, timescale=1000) +├── trak (per track) +│ ├── tkhd (track header, width/height, track_id) +│ └── mdia +│ ├── mdhd (timescale: 90000 for video, 48000 for audio) +│ ├── hdlr ("vide" or "soun") +│ └── minf +│ ├── vmhd/smhd +│ ├── dinf → dref +│ └── stbl +│ ├── stsd → sample entry (avc1/hvc1/mp4a) +│ │ └── codec_private (avcC / hvcC / esds) +│ ├── stts (empty) +│ ├── stsc (empty) +│ ├── stsz (empty) +│ └── stco (empty) +└── mvex + └── trex (per track, default sample description index = 1) +``` +The stbl tables are empty because all timing lives in moof fragments (fragmented MP4). + +--- + +### 11. moof + mdat Generation +**Function:** `build_moof_box()` (line ~817) and `build_fragment_from_sample()` (line ~1103) + +Each `EsSample` becomes exactly **one moof+mdat pair** (CMAF per-sample fragment): + +``` +moof +├── mfhd (sequence_number — globally incrementing) +└── traf + ├── tfhd (track_id, default-base-is-moof flag) + ├── tfdt (version=1, 64-bit base_decode_time) + └── trun (sample_count=1) + ├── data_offset (points past moof into mdat payload) + ├── sample_duration + ├── sample_size + ├── sample_flags (0x02000000 = sync, 0x00010000 = non-sync) + └── sample_composition_time_offset (always 0 in this impl) +mdat +└── [sample bytes — length-prefixed NALUs or raw AAC] +``` +**Timing model:** + +- `base_decode_time` is **accumulation-based**: each sample's `tfdt` value = previous sample's value + previous `sample_duration`. This guarantees strict monotonicity regardless of PTS jitter from SRT transport, preventing MSE "overlapping payload" drops on the player. +- Audio `sample_duration` is always exactly **1024** (in the audio timescale) — the fixed AAC frame size. +- Video `sample_duration` is computed from PTS deltas: `(current_pts_us - last_pts_us) × timescale / 1,000,000` +- Timescale: 90000 for video, actual sample rate (e.g. 48000) for audio +- tfdt uses version 1 (64-bit) to avoid overflow during long streams +- Per-track decode time is tracked in `decode_time_by_track` map (initialized to 0, incremented by each sample's duration) + +**trun flags = 0x000F01:** + +- 0x000001 = data-offset-present +- 0x000100 = sample-duration-present +- 0x000200 = sample-size-present +- 0x000400 = sample-flags-present +- 0x000800 = sample-composition-time-offset-present + +--- + +### 12. MediaFragment → MoQ Object +**Struct:** `MediaFragment` (in `cmaf_segmenter.h`) + +``` +struct MediaFragment { + std::size_t group_id; + std::size_t object_id; + std::string track_name; // e.g. "srt1_video" or "srt1_audio" + std::uint64_t start_time_us; + std::uint64_t duration_us; + bool is_video_keyframe; + PayloadBuffer payload; // .owned_bytes = moof + mdat concatenated +}; +``` +The `FragmentSink` callback pushes each `MediaFragment` into a shared queue. The MoQ session's `drain_queue` loop consumes fragments and writes them as MoQ OBJECT messages on subgroup streams: + +``` +// In moqt_session.cpp drain_queue: +sender.serve(transport_, draft_version, track_alias, send_seq, + object, /*new_subgroup=*/true, /*fin=*/false, payload); +``` +Each fragment's `payload.owned_bytes` (the raw moof+mdat bytes) becomes the **MoQ object payload** — sent as-is on the wire. + +--- + +### 13. MoQ group_id, subgroup_id, and object_id Assignment +**In `build_fragment_from_sample()`** (line ~1113): + +``` +group_id: + - Starts at 0 + - Incremented on each VIDEO KEYFRAME (when fragment_on_keyframe=true) + - All video and audio samples between keyframes share the same group_id + - A new group = new CMAF segment boundary + +subgroup_id: + - Always 0 (single subgroup per group) + +object_id: + - Per-track counter within a group + - Reset to 0 at each new group (on video keyframe) + - Incremented for each sample of that track within the group + - Video and audio have INDEPENDENT object_id sequences +``` +**Example with 30fps video + 48kHz audio (1024-sample AAC frames, ~21ms):** + +``` +Group 0 (starts at keyframe): + video object_id: 0, 1, 2, ... 59 (60 frames @ 30fps = 2 seconds) + audio object_id: 0, 1, 2, ... 92 (~93 AAC frames in 2 seconds) + +Group 1 (next keyframe): + video object_id: 0, 1, 2, ... (reset) + audio object_id: 0, 1, 2, ... (reset) +``` + +--- + +### 14. Frames Per MoQ Object +**Exactly 1 frame (or 1 AAC access unit) per MoQ object.** + +The SRT path creates one moof+mdat per `EsSample`. Each `EsSample` is one complete access unit: + +- **Video:** 1 coded picture (1 frame) +- **Audio:** 1 raw AAC frame (1024 PCM samples ≈ 21.3ms at 48kHz). Even when the encoder packs multiple AAC frames into a single MPEG-TS PES packet, `flush_pes()` splits them into individual `EsSample`s with interpolated timestamps — guaranteeing one access unit per MoQ object. + +This is true CMAF "per-sample" fragmentation — the finest granularity possible. + +--- + +## Comparison: SRT Path vs. Stdin fMP4 Path + +Aspect | SRT Path | Stdin (fragmented MP4) Path +--- | --- | --- +**Input format** | Raw MPEG-TS over SRT | Fragmented MP4 (e.g. `ffmpeg -f mp4 -movflags frag_keyframe+empty_moov pipe:1`) +**Who creates moof+mdat** | moqxr builds it from scratch | FFmpeg (or other tool) creates it; moqxr forwards as-is +**Init segment** | Synthesized from extracted SPS/PPS/VPS/ADTS | Read directly from stdin (ftyp+moov boxes) +**Demuxing** | Full TS demux: PAT→PMT→PES→ES | MP4 box parser: reads top-level `moof` and `mdat` boxes +**Codec conversion** | Annex-B → length-prefixed; ADTS → raw AAC | None needed (already in MP4 format) +**Granularity** | Always 1 sample per moof+mdat | Depends on ffmpeg fragmentation settings (could be N samples per moof) +**Timing** | Accumulation-based (per-track running sum of sample_duration) | Preserved from ffmpeg's trun entries +**group_id** | Incremented on video keyframe | Incremented on video keyframe +**object_id** | Per-track, reset each group | Per-track, reset each group + +``` +STDIN PATH: + ffmpeg → [ftyp+moov] [moof+mdat] [moof+mdat] ... + │ │ + │ └── Forwarded as MoQ Object payload (1:1) + └── Used as init segment + +SRT PATH: + Encoder → SRT → MPEG-TS → PAT/PMT/PES → EsSample + │ + ┌──────────────────────────┘ + ▼ + annexb_to_avcc() / strip_adts() + │ + ▼ + build_moof_box() + build_mdat_box() + │ + ▼ + MoQ Object payload (1 frame per object) +``` + +--- + +## Key Data Structures + +``` +CallerTrackState (per SRT connection): +├── video_track_name / audio_track_name +├── video_track_id / audio_track_id +├── video_timescale (90000) / audio_timescale (48000) +├── group_id (increments on keyframe) +├── first_video_keyframe_seen +├── object_id_by_track (reset per group) +├── last_pts_by_track (for duration calculation) +├── decode_time_by_track (per-track running tfdt, guarantees monotonicity) +├── last_duration_us_by_track (for computing video sample_duration from PTS deltas) +├── moof_sequence (globally incrementing per connection) +└── video_codec (H264 or HEVC) +``` + +--- + +## Codec Discovery Phase +Before streaming begins, the system waits up to 5 seconds for: + +1. First video frame → detect codec type (H.264 vs HEVC) from stream_type or NAL inspection +2. First video keyframe → extract SPS/PPS/VPS → build avcC or hvcC +3. First audio ADTS frame → extract sample rate / channels → build esds + +After discovery: + +- Tracks with no `codec_private` (e.g. audio when feed is video-only) are removed +- Init segment is rebuilt with actual codec parameters +- Catalog is generated and published as MoQ object (group=0, object=0 on "catalog" track) diff --git a/include/openmoq/publisher/cli_options.h b/include/openmoq/publisher/cli_options.h index eea7db4..b8efb9d 100644 --- a/include/openmoq/publisher/cli_options.h +++ b/include/openmoq/publisher/cli_options.h @@ -20,8 +20,16 @@ struct InputSource { std::filesystem::path path; }; +enum class LiveSourceKind { + kAuto, + kStdin, + kSrt, +}; + struct CliOptions { InputSource input_source; + LiveSourceKind live_source = LiveSourceKind::kAuto; + std::optional srt_config_path; std::optional emit_dir; std::optional endpoint; transport::TransportKind transport = transport::TransportKind::kRawQuic; diff --git a/include/openmoq/publisher/cmaf_segmenter.h b/include/openmoq/publisher/cmaf_segmenter.h index 18b3b49..8c287cb 100644 --- a/include/openmoq/publisher/cmaf_segmenter.h +++ b/include/openmoq/publisher/cmaf_segmenter.h @@ -26,6 +26,7 @@ struct MediaFragment { std::uint64_t earliest_presentation_time_us = 0; std::uint8_t sap_type = 0; bool is_video_keyframe = false; // True if this is a video track IDR/keyframe fragment + std::uint64_t creation_time_us = 0; // Wall-clock time when fragment was created (for queue delay measurement) PayloadBuffer payload; }; diff --git a/include/openmoq/publisher/live_srt_config.h b/include/openmoq/publisher/live_srt_config.h new file mode 100644 index 0000000..7b4ece8 --- /dev/null +++ b/include/openmoq/publisher/live_srt_config.h @@ -0,0 +1,46 @@ +#pragma once + +#include +#include +#include +#include +#include + +namespace openmoq::publisher { + +struct SrtSocketConfig { + std::string mode = "caller"; + std::string host; + std::uint16_t port = 0; + std::uint32_t latency_ms = 120; +}; + +struct MpegTsProgramConfig { + bool auto_detect_program = true; + std::optional program_number; + std::optional video_pid; + std::optional audio_pid; +}; + +struct CmafFragmentPolicy { + bool fragment_on_keyframe = true; + bool empty_moov = true; + bool default_base_moof = true; + bool separate_moof_per_track = true; + std::uint32_t target_fragment_duration_ms = 1000; +}; + +struct SrtCallerIngestConfig { + std::string id; + SrtSocketConfig srt; + MpegTsProgramConfig mpegts; + CmafFragmentPolicy cmaf; +}; + +struct LiveSrtConfig { + std::vector srt_callers; +}; + +LiveSrtConfig parse_live_srt_config_file(const std::filesystem::path& path); + +} // namespace openmoq::publisher diff --git a/include/openmoq/publisher/live_srt_ingest.h b/include/openmoq/publisher/live_srt_ingest.h new file mode 100644 index 0000000..deef782 --- /dev/null +++ b/include/openmoq/publisher/live_srt_ingest.h @@ -0,0 +1,63 @@ +#pragma once + +#include +#include +#include +#include +#include +#include + +#include "openmoq/publisher/cmaf_segmenter.h" +#include "openmoq/publisher/mp4_box.h" +#include "openmoq/publisher/transport/publisher_transport.h" + +namespace openmoq::publisher { + +struct LiveSrtCallerRuntimeConfig { + std::string id; + std::string endpoint; + bool fragment_on_keyframe = true; + bool empty_moov = true; + bool default_base_moof = true; + bool separate_moof_per_track = true; + std::uint32_t target_fragment_duration_ms = 1000; + std::uint32_t latency_ms = 120; + bool auto_detect_program = true; + std::uint32_t program_number = 0; + bool has_program_number = false; + std::uint32_t video_pid = 0; + bool has_video_pid = false; + std::uint32_t audio_pid = 0; + bool has_audio_pid = false; +}; + +struct LiveSrtBootstrap { + std::vector tracks; + std::vector init_segment; +}; + +class LiveSrtIngestManager { +public: + using FragmentSink = std::function; + + LiveSrtIngestManager(std::vector callers, + FragmentSink sink, + std::atomic& stop_requested); + + transport::TransportStatus start(); + void join(); + + const LiveSrtBootstrap& bootstrap() const; + + static std::vector build_synthetic_init_segment(const std::vector& tracks); + +private: + struct Impl; + std::vector callers_; + FragmentSink sink_; + std::atomic& stop_requested_; + LiveSrtBootstrap bootstrap_; + std::vector worker_threads_; +}; + +} // namespace openmoq::publisher diff --git a/include/openmoq/publisher/mp4_box.h b/include/openmoq/publisher/mp4_box.h index 2a1650c..87336c5 100644 --- a/include/openmoq/publisher/mp4_box.h +++ b/include/openmoq/publisher/mp4_box.h @@ -39,6 +39,7 @@ struct TrackDescription { std::uint32_t channel_count = 0; std::uint32_t sample_rate = 0; double frame_rate = 0.0; + std::vector codec_private; // avcC, hvcC, or esds box bytes (including box header) }; struct ParsedMp4 { diff --git a/include/openmoq/publisher/publisher_api.h b/include/openmoq/publisher/publisher_api.h index 5315934..89abf73 100644 --- a/include/openmoq/publisher/publisher_api.h +++ b/include/openmoq/publisher/publisher_api.h @@ -53,6 +53,26 @@ struct PublisherStats { std::string last_error; }; +struct LiveSrtCaller { + std::string id; + std::string endpoint; + bool fragment_on_keyframe = true; + bool empty_moov = true; + bool default_base_moof = true; + bool separate_moof_per_track = true; + std::uint32_t target_fragment_duration_ms = 1000; + std::uint32_t latency_ms = 120; + bool auto_detect_program = true; + std::optional program_number; + std::optional video_pid; + std::optional audio_pid; +}; + +struct LiveIngestConfig { + bool use_stdin = false; + std::vector srt_callers; +}; + class Publisher { public: using TransportFactory = std::function(transport::TransportKind)>; @@ -85,6 +105,11 @@ class Publisher { const transport::EndpointConfig& endpoint, const transport::TlsConfig& tls = {}, bool endpoint_alpn_overridden = false) const; + transport::TransportStatus publish_live(const LiveIngestConfig& ingest, + std::istream* stdin_input, + const transport::EndpointConfig& endpoint, + const transport::TlsConfig& tls = {}, + bool endpoint_alpn_overridden = false) const; transport::TransportStatus publish_live_objects(const LiveObjectSource& source, const transport::EndpointConfig& endpoint, const transport::TlsConfig& tls = {}, diff --git a/include/openmoq/publisher/transport/moqt_session.h b/include/openmoq/publisher/transport/moqt_session.h index 0f21f14..1b90b63 100644 --- a/include/openmoq/publisher/transport/moqt_session.h +++ b/include/openmoq/publisher/transport/moqt_session.h @@ -16,6 +16,26 @@ namespace openmoq::publisher::transport { +struct LiveSrtCallerOptions { + std::string id; + std::string endpoint; + bool fragment_on_keyframe = true; + bool empty_moov = true; + bool default_base_moof = true; + bool separate_moof_per_track = true; + std::uint32_t target_fragment_duration_ms = 1000; + std::uint32_t latency_ms = 120; + bool auto_detect_program = true; + std::optional program_number; + std::optional video_pid; + std::optional audio_pid; +}; + +struct LiveIngestOptions { + bool use_stdin = false; + std::vector srt_callers; +}; + class MoqtSession { public: struct PublishStats { @@ -44,6 +64,10 @@ class MoqtSession { TransportStatus publish_live(std::istream& input, openmoq::publisher::DraftVersion draft_version, bool split_cmaf_chunks); + TransportStatus publish_live(const LiveIngestOptions& ingest, + std::istream* stdin_input, + openmoq::publisher::DraftVersion draft_version, + bool split_cmaf_chunks); TransportStatus publish_live_objects(const openmoq::publisher::LiveObjectSource& source, openmoq::publisher::DraftVersion draft_version); TransportStatus close(std::uint64_t application_error_code = 0); diff --git a/src/cli_options.cpp b/src/cli_options.cpp index 72e4621..e1d5748 100644 --- a/src/cli_options.cpp +++ b/src/cli_options.cpp @@ -103,6 +103,22 @@ std::chrono::seconds parse_timeout(std::string_view value) { return std::chrono::seconds(timeout); } +LiveSourceKind parse_live_source(std::string_view value) { + if (value == "auto") { + return LiveSourceKind::kAuto; + } + if (value == "stdin") { + return LiveSourceKind::kStdin; + } + if (value == "srt") { + return LiveSourceKind::kSrt; + } + if (value == "both") { + throw std::runtime_error("--live-source 'both' is not supported; use either 'stdin' or 'srt'"); + } + throw std::runtime_error("unsupported --live-source value: expected auto, stdin, or srt"); +} + } // namespace CliOptions parse_cli_options(int argc, char** argv) { @@ -121,6 +137,10 @@ CliOptions parse_cli_options(int argc, char** argv) { if (argument == "--input") { options.input_source = parse_input_source(require_value("--input")); + } else if (argument == "--live-source") { + options.live_source = parse_live_source(require_value("--live-source")); + } else if (argument == "--srt-config") { + options.srt_config_path = std::filesystem::path(require_value("--srt-config")); } else if (argument == "--transport") { options.transport = parse_transport_kind(require_value("--transport")); } else if (argument == "--endpoint") { @@ -175,10 +195,27 @@ CliOptions parse_cli_options(int argc, char** argv) { } } - if (options.input_source.kind == InputSourceKind::kFile && options.input_source.path.empty()) { + const bool live_source_uses_stdin = + options.live_source == LiveSourceKind::kAuto || + options.live_source == LiveSourceKind::kStdin; + if (live_source_uses_stdin && + options.input_source.kind == InputSourceKind::kFile && + options.input_source.path.empty()) { throw std::runtime_error("missing required --input argument"); } + const bool live_source_uses_srt = + options.live_source == LiveSourceKind::kSrt; + if (live_source_uses_srt && !options.srt_config_path.has_value()) { + throw std::runtime_error("--live-source srt requires --srt-config"); + } + if (live_source_uses_srt && !options.endpoint.has_value()) { + throw std::runtime_error("--live-source srt requires --endpoint"); + } + if (!live_source_uses_srt && options.srt_config_path.has_value()) { + throw std::runtime_error("--srt-config requires --live-source srt"); + } + if (options.endpoint.has_value() && options.endpoint->host.empty()) { throw std::runtime_error("--alpn and --sni require --endpoint to be provided first"); } @@ -202,7 +239,8 @@ CliOptions parse_cli_options(int argc, char** argv) { std::string build_usage(const char* argv0) { return std::string("Usage: ") + argv0 + - " --input [--transport raw|webtransport] [--draft 14|16|17|18] [--namespace ] [--forward 0|1] [--timeout ]" + " --input [--live-source auto|stdin|srt] [--srt-config ]" + " [--transport raw|webtransport] [--draft 14|16|17|18] [--namespace ] [--forward 0|1] [--timeout ]" " [--publish-catalog] [--sap] [--msf-timeline] [--coalesce-cmaf-chunks] [--paced] [--loop] [--dump-plan] [--emit-dir ]" " [--endpoint host:port|moqt://host:port/path|https://host:port/path] [--alpn value] [--sni value]" " [--cert file] [--key file] [--ca file] [--insecure]"; diff --git a/src/live_srt_config.cpp b/src/live_srt_config.cpp new file mode 100644 index 0000000..0d21505 --- /dev/null +++ b/src/live_srt_config.cpp @@ -0,0 +1,409 @@ +#include "openmoq/publisher/live_srt_config.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace openmoq::publisher { + +namespace { + +struct JsonValue; +using JsonObject = std::unordered_map; +using JsonArray = std::vector; + +// Use unique_ptr wrappers for recursive types to avoid incomplete-type errors on GCC 11. +struct JsonValue { + using Value = std::variant, std::unique_ptr>; + Value value; + + // Convenience constructors so existing code using JsonValue{...} still works. + JsonValue() : value(nullptr) {} + JsonValue(std::nullptr_t) : value(nullptr) {} + JsonValue(bool b) : value(b) {} + JsonValue(double d) : value(d) {} + JsonValue(std::string s) : value(std::move(s)) {} + JsonValue(JsonArray arr) : value(std::make_unique(std::move(arr))) {} + JsonValue(JsonObject obj) : value(std::make_unique(std::move(obj))) {} + + // Copy support (needed for vector/map operations) + JsonValue(const JsonValue& other); + JsonValue& operator=(const JsonValue& other); + JsonValue(JsonValue&&) = default; + JsonValue& operator=(JsonValue&&) = default; + ~JsonValue() = default; +}; + +JsonValue::JsonValue(const JsonValue& other) { + struct CopyVisitor { + JsonValue::Value operator()(std::nullptr_t) const { return nullptr; } + JsonValue::Value operator()(bool b) const { return b; } + JsonValue::Value operator()(double d) const { return d; } + JsonValue::Value operator()(const std::string& s) const { return s; } + JsonValue::Value operator()(const std::unique_ptr& a) const { + return a ? std::make_unique(*a) : std::make_unique(); + } + JsonValue::Value operator()(const std::unique_ptr& o) const { + return o ? std::make_unique(*o) : std::make_unique(); + } + }; + value = std::visit(CopyVisitor{}, other.value); +} + +JsonValue& JsonValue::operator=(const JsonValue& other) { + if (this != &other) { + JsonValue tmp(other); + value = std::move(tmp.value); + } + return *this; +} + +class JsonParser { +public: + explicit JsonParser(std::string input) + : input_(std::move(input)) {} + + JsonValue parse() { + skip_ws(); + JsonValue root = parse_value(); + skip_ws(); + if (!eof()) { + throw std::runtime_error("unexpected trailing JSON content"); + } + return root; + } + +private: + JsonValue parse_value() { + if (eof()) { + throw std::runtime_error("unexpected end of JSON input"); + } + + const char ch = peek(); + if (ch == '{') { + return JsonValue{parse_object()}; + } + if (ch == '[') { + return JsonValue{parse_array()}; + } + if (ch == '"') { + return JsonValue{parse_string()}; + } + if (ch == 't') { + consume_literal("true"); + return JsonValue{true}; + } + if (ch == 'f') { + consume_literal("false"); + return JsonValue{false}; + } + if (ch == 'n') { + consume_literal("null"); + return JsonValue{nullptr}; + } + if (ch == '-' || std::isdigit(static_cast(ch)) != 0) { + return JsonValue{parse_number()}; + } + throw std::runtime_error("unsupported JSON token"); + } + + JsonObject parse_object() { + expect('{'); + skip_ws(); + JsonObject object; + if (consume_if('}')) { + return object; + } + + while (true) { + skip_ws(); + const std::string key = parse_string(); + skip_ws(); + expect(':'); + skip_ws(); + object.emplace(key, parse_value()); + skip_ws(); + if (consume_if('}')) { + break; + } + expect(','); + } + return object; + } + + JsonArray parse_array() { + expect('['); + skip_ws(); + JsonArray array; + if (consume_if(']')) { + return array; + } + + while (true) { + skip_ws(); + array.push_back(parse_value()); + skip_ws(); + if (consume_if(']')) { + break; + } + expect(','); + } + return array; + } + + std::string parse_string() { + expect('"'); + std::string out; + while (!eof()) { + const char ch = take(); + if (ch == '"') { + return out; + } + if (ch == '\\') { + if (eof()) { + throw std::runtime_error("unterminated JSON escape"); + } + const char esc = take(); + switch (esc) { + case '"': out.push_back('"'); break; + case '\\': out.push_back('\\'); break; + case '/': out.push_back('/'); break; + case 'b': out.push_back('\b'); break; + case 'f': out.push_back('\f'); break; + case 'n': out.push_back('\n'); break; + case 'r': out.push_back('\r'); break; + case 't': out.push_back('\t'); break; + default: + throw std::runtime_error("unsupported JSON escape sequence"); + } + continue; + } + out.push_back(ch); + } + throw std::runtime_error("unterminated JSON string"); + } + + double parse_number() { + const std::size_t start = pos_; + if (peek() == '-') { + take(); + } + if (eof()) { + throw std::runtime_error("invalid JSON number"); + } + if (peek() == '0') { + take(); + } else { + while (!eof() && std::isdigit(static_cast(peek())) != 0) { + take(); + } + } + if (!eof() && peek() == '.') { + take(); + if (eof() || std::isdigit(static_cast(peek())) == 0) { + throw std::runtime_error("invalid JSON number fraction"); + } + while (!eof() && std::isdigit(static_cast(peek())) != 0) { + take(); + } + } + if (!eof() && (peek() == 'e' || peek() == 'E')) { + take(); + if (!eof() && (peek() == '+' || peek() == '-')) { + take(); + } + if (eof() || std::isdigit(static_cast(peek())) == 0) { + throw std::runtime_error("invalid JSON exponent"); + } + while (!eof() && std::isdigit(static_cast(peek())) != 0) { + take(); + } + } + + return std::stod(input_.substr(start, pos_ - start)); + } + + void consume_literal(std::string_view literal) { + for (const char ch : literal) { + if (eof() || take() != ch) { + throw std::runtime_error("invalid JSON literal"); + } + } + } + + void skip_ws() { + while (!eof() && std::isspace(static_cast(peek())) != 0) { + ++pos_; + } + } + + void expect(char ch) { + if (eof() || take() != ch) { + throw std::runtime_error("unexpected JSON token"); + } + } + + bool consume_if(char ch) { + if (!eof() && peek() == ch) { + ++pos_; + return true; + } + return false; + } + + char peek() const { + return input_[pos_]; + } + + char take() { + return input_[pos_++]; + } + + bool eof() const { + return pos_ >= input_.size(); + } + + std::string input_; + std::size_t pos_ = 0; +}; + +const JsonObject& expect_object(const JsonValue& value, std::string_view field_name) { + const auto* ptr = std::get_if>(&value.value); + if (ptr == nullptr || *ptr == nullptr) { + throw std::runtime_error(std::string(field_name) + " must be an object"); + } + return **ptr; +} + +const JsonArray& expect_array(const JsonValue& value, std::string_view field_name) { + const auto* ptr = std::get_if>(&value.value); + if (ptr == nullptr || *ptr == nullptr) { + throw std::runtime_error(std::string(field_name) + " must be an array"); + } + return **ptr; +} + +std::string expect_string(const JsonObject& object, std::string_view key) { + const auto it = object.find(std::string(key)); + if (it == object.end()) { + throw std::runtime_error("missing JSON key: " + std::string(key)); + } + const auto* str = std::get_if(&it->second.value); + if (str == nullptr) { + throw std::runtime_error("JSON key must be a string: " + std::string(key)); + } + return *str; +} + +bool read_bool_or_default(const JsonObject& object, std::string_view key, bool default_value) { + const auto it = object.find(std::string(key)); + if (it == object.end()) { + return default_value; + } + const auto* value = std::get_if(&it->second.value); + if (value == nullptr) { + throw std::runtime_error("JSON key must be a bool: " + std::string(key)); + } + return *value; +} + +std::optional read_optional_u32(const JsonObject& object, std::string_view key) { + const auto it = object.find(std::string(key)); + if (it == object.end()) { + return std::nullopt; + } + if (std::holds_alternative(it->second.value)) { + return std::nullopt; + } + const auto* number = std::get_if(&it->second.value); + if (number == nullptr) { + throw std::runtime_error("JSON key must be a number or null: " + std::string(key)); + } + if (*number < 0.0 || *number > static_cast(std::numeric_limits::max())) { + throw std::runtime_error("JSON key out of range: " + std::string(key)); + } + return static_cast(*number); +} + +std::uint32_t read_u32_or_default(const JsonObject& object, std::string_view key, std::uint32_t default_value) { + const auto value = read_optional_u32(object, key); + return value.has_value() ? *value : default_value; +} + +} // namespace + +LiveSrtConfig parse_live_srt_config_file(const std::filesystem::path& path) { + std::ifstream file(path); + if (!file.is_open()) { + throw std::runtime_error("failed to open SRT config file: " + path.string()); + } + + std::string json_text((std::istreambuf_iterator(file)), std::istreambuf_iterator()); + JsonParser parser(std::move(json_text)); + const JsonValue root = parser.parse(); + const JsonObject& root_obj = expect_object(root, "root"); + + const auto callers_it = root_obj.find("srt_callers"); + if (callers_it == root_obj.end()) { + throw std::runtime_error("missing JSON key: srt_callers"); + } + const JsonArray& callers = expect_array(callers_it->second, "srt_callers"); + + LiveSrtConfig config; + config.srt_callers.reserve(callers.size()); + + for (const JsonValue& caller_value : callers) { + const JsonObject& caller_obj = expect_object(caller_value, "srt_callers[]"); + SrtCallerIngestConfig caller; + caller.id = expect_string(caller_obj, "id"); + + const auto srt_it = caller_obj.find("srt"); + if (srt_it == caller_obj.end()) { + throw std::runtime_error("missing JSON key: srt"); + } + const JsonObject& srt_obj = expect_object(srt_it->second, "srt"); + caller.srt.mode = expect_string(srt_obj, "mode"); + if (caller.srt.mode != "caller") { + throw std::runtime_error("unsupported srt.mode '" + caller.srt.mode + "': only 'caller' is supported"); + } + caller.srt.host = expect_string(srt_obj, "host"); + const auto port = read_optional_u32(srt_obj, "port"); + if (!port.has_value() || *port == 0 || *port > 65535) { + throw std::runtime_error("srt.port must be between 1 and 65535"); + } + caller.srt.port = static_cast(*port); + caller.srt.latency_ms = read_u32_or_default(srt_obj, "latency_ms", 120); + + const auto mpegts_it = caller_obj.find("mpegts"); + if (mpegts_it != caller_obj.end()) { + const JsonObject& mpegts_obj = expect_object(mpegts_it->second, "mpegts"); + caller.mpegts.auto_detect_program = read_bool_or_default(mpegts_obj, "auto_detect_program", true); + caller.mpegts.program_number = read_optional_u32(mpegts_obj, "program_number"); + caller.mpegts.video_pid = read_optional_u32(mpegts_obj, "video_pid"); + caller.mpegts.audio_pid = read_optional_u32(mpegts_obj, "audio_pid"); + } + + const auto cmaf_it = caller_obj.find("cmaf"); + if (cmaf_it != caller_obj.end()) { + const JsonObject& cmaf_obj = expect_object(cmaf_it->second, "cmaf"); + caller.cmaf.fragment_on_keyframe = read_bool_or_default(cmaf_obj, "fragment_on_keyframe", true); + caller.cmaf.empty_moov = read_bool_or_default(cmaf_obj, "empty_moov", true); + caller.cmaf.default_base_moof = read_bool_or_default(cmaf_obj, "default_base_moof", true); + caller.cmaf.separate_moof_per_track = read_bool_or_default(cmaf_obj, "separate_moof_per_track", true); + caller.cmaf.target_fragment_duration_ms = + read_u32_or_default(cmaf_obj, "target_fragment_duration_ms", 1000); + } + + config.srt_callers.push_back(std::move(caller)); + } + + return config; +} + +} // namespace openmoq::publisher diff --git a/src/live_srt_ingest.cpp b/src/live_srt_ingest.cpp new file mode 100644 index 0000000..42731b4 --- /dev/null +++ b/src/live_srt_ingest.cpp @@ -0,0 +1,1820 @@ +#include "openmoq/publisher/live_srt_ingest.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#if defined(OPENMOQ_HAS_SRT) +#include +#include +#include +#include +#include +#endif + +namespace openmoq::publisher { +namespace { + +std::uint16_t read_be16(std::span bytes, std::size_t offset) { + return static_cast((static_cast(bytes[offset]) << 8U) | bytes[offset + 1]); +} + +std::uint32_t read_be24(std::span bytes, std::size_t offset) { + return (static_cast(bytes[offset]) << 16U) | + (static_cast(bytes[offset + 1]) << 8U) | + static_cast(bytes[offset + 2]); +} + +std::uint32_t read_be32(std::span bytes, std::size_t offset) { + return (static_cast(bytes[offset]) << 24U) | + (static_cast(bytes[offset + 1]) << 16U) | + (static_cast(bytes[offset + 2]) << 8U) | + static_cast(bytes[offset + 3]); +} + +void append_be16(std::vector& out, std::uint16_t value) { + out.push_back(static_cast((value >> 8U) & 0xFFU)); + out.push_back(static_cast(value & 0xFFU)); +} + +void append_be32(std::vector& out, std::uint32_t value) { + out.push_back(static_cast((value >> 24U) & 0xFFU)); + out.push_back(static_cast((value >> 16U) & 0xFFU)); + out.push_back(static_cast((value >> 8U) & 0xFFU)); + out.push_back(static_cast(value & 0xFFU)); +} + +void append_be64(std::vector& out, std::uint64_t value) { + for (int i = 7; i >= 0; --i) { + out.push_back(static_cast((value >> (i * 8U)) & 0xFFU)); + } +} + +void append_ascii(std::vector& out, std::string_view text) { + out.insert(out.end(), text.begin(), text.end()); +} + +std::vector make_box(std::string_view type, std::span payload) { + std::vector out; + out.reserve(8 + payload.size()); + append_be32(out, static_cast(8 + payload.size())); + append_ascii(out, type); + out.insert(out.end(), payload.begin(), payload.end()); + return out; +} + +std::vector make_full_box(std::string_view type, + std::uint8_t version, + std::uint32_t flags, + std::span payload) { + std::vector body; + body.reserve(4 + payload.size()); + body.push_back(version); + body.push_back(static_cast((flags >> 16U) & 0xFFU)); + body.push_back(static_cast((flags >> 8U) & 0xFFU)); + body.push_back(static_cast(flags & 0xFFU)); + body.insert(body.end(), payload.begin(), payload.end()); + return make_box(type, body); +} + +std::vector build_ftyp_box() { + std::vector payload; + append_ascii(payload, "isom"); + append_be32(payload, 0x00000200); + append_ascii(payload, "isom"); + append_ascii(payload, "iso6"); + append_ascii(payload, "mp41"); + return make_box("ftyp", payload); +} + +std::vector build_mvhd_box() { + std::vector payload; + append_be32(payload, 0); + append_be32(payload, 0); + append_be32(payload, 1000); + append_be32(payload, 0); + append_be32(payload, 0x00010000); + append_be16(payload, 0x0100); + append_be16(payload, 0); + payload.insert(payload.end(), 8, 0); + append_be32(payload, 0x00010000); + append_be32(payload, 0); + append_be32(payload, 0); + append_be32(payload, 0); + append_be32(payload, 0x00010000); + append_be32(payload, 0); + append_be32(payload, 0); + append_be32(payload, 0); + append_be32(payload, 0x40000000); + payload.insert(payload.end(), 24, 0); + append_be32(payload, 0xFFFFFFFF); + return make_full_box("mvhd", 0, 0, payload); +} + +std::vector build_tkhd_box(const TrackDescription& track) { + std::vector payload; + append_be32(payload, 0); // creation_time + append_be32(payload, 0); // modification_time + append_be32(payload, track.track_id); // track_ID + append_be32(payload, 0); // reserved + append_be32(payload, 0); // duration + append_be32(payload, 0); // reserved[0] + append_be32(payload, 0); // reserved[1] + append_be16(payload, 0); // layer + append_be16(payload, 0); // alternate_group + append_be16(payload, track.handler_type == "soun" ? 0x0100 : 0); // volume + append_be16(payload, 0); // reserved + // matrix (identity) + append_be32(payload, 0x00010000); + append_be32(payload, 0); + append_be32(payload, 0); + append_be32(payload, 0); + append_be32(payload, 0x00010000); + append_be32(payload, 0); + append_be32(payload, 0); + append_be32(payload, 0); + append_be32(payload, 0x40000000); + // width and height (16.16 fixed point) + append_be32(payload, track.width << 16U); + append_be32(payload, track.height << 16U); + return make_full_box("tkhd", 0, 0x000007, payload); +} + +std::vector build_mdhd_box(const TrackDescription& track) { + std::vector payload; + append_be32(payload, 0); + append_be32(payload, 0); + append_be32(payload, track.timescale == 0 ? 90000 : track.timescale); + append_be32(payload, 0); + append_be16(payload, 0x55C4); + append_be16(payload, 0); + return make_full_box("mdhd", 0, 0, payload); +} + +std::vector build_hdlr_box(const TrackDescription& track) { + std::vector payload; + append_be32(payload, 0); + append_ascii(payload, track.handler_type == "soun" ? "soun" : "vide"); + payload.insert(payload.end(), 12, 0); + const std::string name = track.track_name; + payload.insert(payload.end(), name.begin(), name.end()); + payload.push_back(0); + return make_full_box("hdlr", 0, 0, payload); +} + +std::vector build_stsd_box(const TrackDescription& track) { + std::vector sample_entry; + sample_entry.insert(sample_entry.end(), 6, 0); + append_be16(sample_entry, 1); + if (track.handler_type == "soun") { + sample_entry.insert(sample_entry.end(), 8, 0); + append_be16(sample_entry, static_cast(track.channel_count == 0 ? 2 : track.channel_count)); + append_be16(sample_entry, 16); + append_be16(sample_entry, 0); + append_be16(sample_entry, 0); + append_be32(sample_entry, (track.sample_rate == 0 ? 48000 : track.sample_rate) << 16U); + } else { + sample_entry.insert(sample_entry.end(), 16, 0); + append_be16(sample_entry, static_cast(track.width == 0 ? 1920 : track.width)); + append_be16(sample_entry, static_cast(track.height == 0 ? 1080 : track.height)); + append_be32(sample_entry, 0x00480000); + append_be32(sample_entry, 0x00480000); + append_be32(sample_entry, 0); + append_be16(sample_entry, 1); + sample_entry.insert(sample_entry.end(), 32, 0); + append_be16(sample_entry, 24); + append_be16(sample_entry, 0xFFFF); + } + + // Append codec-specific box (avcC, hvcC, esds) inside the sample entry if available + if (!track.codec_private.empty()) { + sample_entry.insert(sample_entry.end(), track.codec_private.begin(), track.codec_private.end()); + } + + std::vector sample_entry_box; + append_be32(sample_entry_box, static_cast(8 + sample_entry.size())); + append_ascii(sample_entry_box, track.sample_entry_type.empty() ? (track.handler_type == "soun" ? "mp4a" : "avc1") + : track.sample_entry_type.substr(0, 4)); + sample_entry_box.insert(sample_entry_box.end(), sample_entry.begin(), sample_entry.end()); + + std::vector payload; + append_be32(payload, 1); + payload.insert(payload.end(), sample_entry_box.begin(), sample_entry_box.end()); + return make_full_box("stsd", 0, 0, payload); +} + +std::vector build_empty_time_table(std::string_view type) { + std::vector payload; + append_be32(payload, 0); + return make_full_box(type, 0, 0, payload); +} + +std::vector build_stsz_box() { + std::vector payload; + append_be32(payload, 0); + append_be32(payload, 0); + return make_full_box("stsz", 0, 0, payload); +} + +std::vector build_stco_box() { + std::vector payload; + append_be32(payload, 0); + return make_full_box("stco", 0, 0, payload); +} + +std::vector build_dinf_box() { + std::vector url_payload; + auto url = make_full_box("url ", 0, 1, url_payload); + std::vector dref_payload; + append_be32(dref_payload, 1); + dref_payload.insert(dref_payload.end(), url.begin(), url.end()); + auto dref = make_full_box("dref", 0, 0, dref_payload); + return make_box("dinf", dref); +} + +std::vector build_minf_box(const TrackDescription& track) { + std::vector vmhd_payload; + append_be16(vmhd_payload, 0); + append_be16(vmhd_payload, 0); + append_be16(vmhd_payload, 0); + append_be16(vmhd_payload, 0); + + std::vector smhd_payload; + append_be16(smhd_payload, 0); + append_be16(smhd_payload, 0); + + std::vector stbl_payload; + auto stsd = build_stsd_box(track); + auto stts = build_empty_time_table("stts"); + auto stsc = build_empty_time_table("stsc"); + auto stsz = build_stsz_box(); + auto stco = build_stco_box(); + stbl_payload.insert(stbl_payload.end(), stsd.begin(), stsd.end()); + stbl_payload.insert(stbl_payload.end(), stts.begin(), stts.end()); + stbl_payload.insert(stbl_payload.end(), stsc.begin(), stsc.end()); + stbl_payload.insert(stbl_payload.end(), stsz.begin(), stsz.end()); + stbl_payload.insert(stbl_payload.end(), stco.begin(), stco.end()); + auto stbl = make_box("stbl", stbl_payload); + + auto dinf = build_dinf_box(); + std::vector minf_payload; + if (track.handler_type == "soun") { + auto smhd = make_full_box("smhd", 0, 0, smhd_payload); + minf_payload.insert(minf_payload.end(), smhd.begin(), smhd.end()); + } else { + auto vmhd = make_full_box("vmhd", 0, 1, vmhd_payload); + minf_payload.insert(minf_payload.end(), vmhd.begin(), vmhd.end()); + } + minf_payload.insert(minf_payload.end(), dinf.begin(), dinf.end()); + minf_payload.insert(minf_payload.end(), stbl.begin(), stbl.end()); + return make_box("minf", minf_payload); +} + +std::vector build_trak_box(const TrackDescription& track) { + auto tkhd = build_tkhd_box(track); + auto mdhd = build_mdhd_box(track); + auto hdlr = build_hdlr_box(track); + auto minf = build_minf_box(track); + + std::vector mdia_payload; + mdia_payload.insert(mdia_payload.end(), mdhd.begin(), mdhd.end()); + mdia_payload.insert(mdia_payload.end(), hdlr.begin(), hdlr.end()); + mdia_payload.insert(mdia_payload.end(), minf.begin(), minf.end()); + auto mdia = make_box("mdia", mdia_payload); + + std::vector trak_payload; + trak_payload.insert(trak_payload.end(), tkhd.begin(), tkhd.end()); + trak_payload.insert(trak_payload.end(), mdia.begin(), mdia.end()); + return make_box("trak", trak_payload); +} + +std::vector build_mvex_box(const std::vector& tracks) { + std::vector payload; + for (const auto& track : tracks) { + std::vector trex_payload; + append_be32(trex_payload, track.track_id); + append_be32(trex_payload, 1); + append_be32(trex_payload, 0); + append_be32(trex_payload, 0); + append_be32(trex_payload, 0); + auto trex = make_full_box("trex", 0, 0, trex_payload); + payload.insert(payload.end(), trex.begin(), trex.end()); + } + return make_box("mvex", payload); +} + +std::vector build_init_segment_from_tracks(const std::vector& tracks) { + auto ftyp = build_ftyp_box(); + auto mvhd = build_mvhd_box(); + auto mvex = build_mvex_box(tracks); + + std::vector moov_payload; + moov_payload.insert(moov_payload.end(), mvhd.begin(), mvhd.end()); + for (const auto& track : tracks) { + auto trak = build_trak_box(track); + moov_payload.insert(moov_payload.end(), trak.begin(), trak.end()); + } + moov_payload.insert(moov_payload.end(), mvex.begin(), mvex.end()); + + auto moov = make_box("moov", moov_payload); + std::vector init; + init.insert(init.end(), ftyp.begin(), ftyp.end()); + init.insert(init.end(), moov.begin(), moov.end()); + return init; +} + +bool h264_annexb_has_idr(std::span payload) { + for (std::size_t i = 0; i + 4 < payload.size(); ++i) { + if (payload[i] == 0x00 && payload[i + 1] == 0x00 && + ((payload[i + 2] == 0x01) || (payload[i + 2] == 0x00 && payload[i + 3] == 0x01))) { + const std::size_t nal_offset = payload[i + 2] == 0x01 ? i + 3 : i + 4; + if (nal_offset < payload.size()) { + const std::uint8_t nal_type = static_cast(payload[nal_offset] & 0x1FU); + if (nal_type == 5) { + return true; + } + } + } + } + return false; +} + +std::vector annexb_to_avcc(std::span payload) { + std::vector out; + std::size_t i = 0; + while (i + 3 < payload.size()) { + std::size_t start = std::string::npos; + std::size_t sc_len = 0; + for (; i + 3 < payload.size(); ++i) { + if (payload[i] == 0x00 && payload[i + 1] == 0x00 && payload[i + 2] == 0x01) { + start = i + 3; + sc_len = 3; + break; + } + if (i + 4 < payload.size() && payload[i] == 0x00 && payload[i + 1] == 0x00 && + payload[i + 2] == 0x00 && payload[i + 3] == 0x01) { + start = i + 4; + sc_len = 4; + break; + } + } + if (start == std::string::npos || start >= payload.size()) { + break; + } + std::size_t next = start; + while (next + 3 < payload.size()) { + if (payload[next] == 0x00 && payload[next + 1] == 0x00 && + (payload[next + 2] == 0x01 || + (next + 3 < payload.size() && payload[next + 2] == 0x00 && payload[next + 3] == 0x01))) { + break; + } + ++next; + } + if (next + 3 >= payload.size()) { + next = payload.size(); + } + const std::size_t nal_size = next - start; + if (nal_size == 0) { + i = next; + continue; + } + append_be32(out, static_cast(nal_size)); + out.insert(out.end(), payload.begin() + static_cast(start), + payload.begin() + static_cast(start + nal_size)); + i = next; + if (sc_len == 0) { + break; + } + } + if (out.empty()) { + return std::vector(payload.begin(), payload.end()); + } + return out; +} + +struct EsSample { + bool is_video = false; + std::uint64_t pts90k = 0; + std::uint8_t stream_type = 0; + std::vector payload; + bool keyframe = false; + // For audio from ADTS: first 9 bytes of the original ADTS header (for codec discovery). + // Empty for video or non-ADTS audio, or after the first frame in a multi-frame PES. + std::array adts_header{}; + std::uint8_t adts_header_len = 0; +}; + +enum class VideoCodec { + kUnknown, + kH264, + kHevc, +}; + +struct CallerTrackState { + std::string video_track_name; + std::string audio_track_name; + std::uint32_t video_track_id = 0; + std::uint32_t audio_track_id = 0; + std::uint32_t video_timescale = 90000; + std::uint32_t audio_timescale = 48000; + std::size_t group_id = 0; + bool first_video_keyframe_seen = false; + std::map object_id_by_track; + std::map last_pts_by_track; + std::map last_duration_us_by_track; + std::map decode_time_by_track; + std::uint32_t moof_sequence = 1; + VideoCodec video_codec = VideoCodec::kUnknown; + // Shared PTS origin (90 kHz) for A/V timeline alignment. + // Set once from the first sample received (video or audio). + std::optional base_pts90k; +}; + +VideoCodec detect_video_codec_from_stream_type(std::uint8_t stream_type) { + if (stream_type == 0x1B || stream_type == 0x02) { + return VideoCodec::kH264; + } + if (stream_type == 0x24) { + return VideoCodec::kHevc; + } + return VideoCodec::kUnknown; +} + +VideoCodec detect_video_codec_from_annexb(std::span payload) { + bool saw_h264_marker = false; + for (std::size_t i = 0; i + 4 < payload.size(); ++i) { + if (payload[i] == 0x00 && payload[i + 1] == 0x00 && + (payload[i + 2] == 0x01 || + (i + 4 < payload.size() && payload[i + 2] == 0x00 && payload[i + 3] == 0x01))) { + const std::size_t nal_offset = payload[i + 2] == 0x01 ? i + 3 : i + 4; + if (nal_offset >= payload.size()) { + continue; + } + const std::uint8_t h264_type = static_cast(payload[nal_offset] & 0x1FU); + if (h264_type == 7 || h264_type == 8 || h264_type == 5) { + saw_h264_marker = true; + } + const std::uint8_t hevc_type = static_cast((payload[nal_offset] >> 1U) & 0x3FU); + if (hevc_type == 32 || hevc_type == 33 || hevc_type == 34 || + hevc_type == 19 || hevc_type == 20 || hevc_type == 21) { + return VideoCodec::kHevc; + } + } + } + return saw_h264_marker ? VideoCodec::kH264 : VideoCodec::kUnknown; +} + +bool hevc_annexb_has_irap(std::span payload) { + for (std::size_t i = 0; i + 4 < payload.size(); ++i) { + if (payload[i] == 0x00 && payload[i + 1] == 0x00 && + (payload[i + 2] == 0x01 || + (i + 4 < payload.size() && payload[i + 2] == 0x00 && payload[i + 3] == 0x01))) { + const std::size_t nal_offset = payload[i + 2] == 0x01 ? i + 3 : i + 4; + if (nal_offset < payload.size()) { + const std::uint8_t nal_type = static_cast((payload[nal_offset] >> 1U) & 0x3FU); + if (nal_type >= 16 && nal_type <= 23) { + return true; + } + } + } + } + return false; +} + +// Extract individual NAL units from AnnexB-formatted H.264/HEVC bitstream. +std::vector> extract_nal_units(std::span payload) { + std::vector> nals; + std::size_t i = 0; + while (i + 3 < payload.size()) { + std::size_t start = std::string::npos; + for (; i + 3 < payload.size(); ++i) { + if (payload[i] == 0x00 && payload[i + 1] == 0x00) { + if (payload[i + 2] == 0x01) { + start = i + 3; + i = start; + break; + } + if (i + 3 < payload.size() && payload[i + 2] == 0x00 && payload[i + 3] == 0x01) { + start = i + 4; + i = start; + break; + } + } + } + if (start == std::string::npos || start >= payload.size()) { + break; + } + // Find next start code or end of data + std::size_t end = start; + while (end + 3 < payload.size()) { + if (payload[end] == 0x00 && payload[end + 1] == 0x00 && + (payload[end + 2] == 0x01 || + (end + 3 < payload.size() && payload[end + 2] == 0x00 && payload[end + 3] == 0x01))) { + break; + } + ++end; + } + if (end == start && end + 3 >= payload.size()) { + end = payload.size(); + } else if (end == start) { + continue; + } + // Trim trailing zeros from NAL unit + std::size_t trimmed_end = end; + while (trimmed_end > start && payload[trimmed_end - 1] == 0x00) { + --trimmed_end; + } + if (trimmed_end > start) { + nals.emplace_back(payload.begin() + static_cast(start), + payload.begin() + static_cast(trimmed_end)); + } + i = end; + } + return nals; +} + +// Build an avcC box (AVCDecoderConfigurationRecord) from SPS and PPS NAL units. +// Returns the complete box including the 4-byte type header. +std::vector build_avcc_box(const std::vector& sps, + const std::vector& pps) { + if (sps.size() < 4 || pps.empty()) { + return {}; + } + // AVCDecoderConfigurationRecord + std::vector record; + record.push_back(1); // configurationVersion + record.push_back(sps[1]); // AVCProfileIndication + record.push_back(sps[2]); // profile_compatibility + record.push_back(sps[3]); // AVCLevelIndication + record.push_back(0xFF); // lengthSizeMinusOne = 3 (4-byte NAL lengths) | reserved 6 bits + record.push_back(static_cast(0xE0U | 1U)); // numOfSequenceParameterSets = 1 | reserved 3 bits + append_be16(record, static_cast(sps.size())); + record.insert(record.end(), sps.begin(), sps.end()); + record.push_back(1); // numOfPictureParameterSets + append_be16(record, static_cast(pps.size())); + record.insert(record.end(), pps.begin(), pps.end()); + // Wrap in box + return make_box("avcC", record); +} + +// Extract VPS, SPS, PPS from AnnexB HEVC keyframe. +struct HevcParamSets { + std::vector vps; + std::vector sps; + std::vector pps; +}; + +HevcParamSets extract_hevc_param_sets(std::span payload) { + HevcParamSets params; + auto nals = extract_nal_units(payload); + for (const auto& nal : nals) { + if (nal.size() < 2) continue; + const std::uint8_t nal_type = static_cast((nal[0] >> 1U) & 0x3FU); + if (nal_type == 32 && params.vps.empty()) { // VPS + params.vps = nal; + } else if (nal_type == 33 && params.sps.empty()) { // SPS + params.sps = nal; + } else if (nal_type == 34 && params.pps.empty()) { // PPS + params.pps = nal; + } + if (!params.vps.empty() && !params.sps.empty() && !params.pps.empty()) break; + } + return params; +} + +// Minimal RBSP bitstream reader used only for SPS dimension extraction. +class RbspBitReader { +public: + RbspBitReader(const std::uint8_t* data, std::size_t size) : data_(data), size_(size) {} + + bool read_bit(std::uint8_t& out) { + if (byte_pos_ >= size_) return false; + out = static_cast((data_[byte_pos_] >> (7U - bit_pos_)) & 0x01U); + if (++bit_pos_ == 8U) { bit_pos_ = 0; ++byte_pos_; } + return true; + } + + bool skip_bits(std::size_t count) { + for (std::size_t k = 0; k < count; ++k) { + std::uint8_t b = 0; + if (!read_bit(b)) return false; + } + return true; + } + + bool read_ue(std::uint32_t& out) { + int zeros = 0; + std::uint8_t bit = 0; + while (zeros < 32) { + if (!read_bit(bit)) return false; + if (bit != 0) break; + ++zeros; + } + if (zeros == 32) return false; + std::uint32_t suffix = 0; + for (int k = 0; k < zeros; ++k) { + if (!read_bit(bit)) return false; + suffix = (suffix << 1U) | bit; + } + out = (1U << zeros) - 1U + suffix; + return true; + } + + bool skip_ue() { std::uint32_t v = 0; return read_ue(v); } + + bool read_se(std::int32_t& out) { + std::uint32_t ue = 0; + if (!read_ue(ue)) return false; + out = (ue % 2U == 0U) ? -static_cast(ue / 2U) + : static_cast((ue + 1U) / 2U); + return true; + } + + bool skip_se() { std::int32_t v = 0; return read_se(v); } + +private: + const std::uint8_t* data_; + std::size_t size_; + std::size_t byte_pos_ = 0; + std::uint8_t bit_pos_ = 0; +}; + +// Parse {width, height} from a raw H.264 SPS NAL unit (includes 1-byte NAL header). +// Returns {0, 0} on any parse failure. +std::pair +parse_h264_dimensions(const std::vector& sps_nal) { + if (sps_nal.size() < 5) return {0, 0}; + const std::uint8_t profile_idc = sps_nal[1]; + RbspBitReader r(sps_nal.data() + 4, sps_nal.size() - 4); + + if (!r.skip_ue()) return {0, 0}; // seq_parameter_set_id + + std::uint32_t chroma_format_idc = 1; + static constexpr std::uint8_t kHighProfiles[] = { + 100, 110, 122, 244, 44, 83, 86, 118, 128, 138, 139, 134, 135 + }; + for (std::uint8_t hp : kHighProfiles) { + if (profile_idc != hp) continue; + if (!r.read_ue(chroma_format_idc)) return {0, 0}; + if (chroma_format_idc == 3 && !r.skip_bits(1)) return {0, 0}; + if (!r.skip_ue()) return {0, 0}; // bit_depth_luma_minus8 + if (!r.skip_ue()) return {0, 0}; // bit_depth_chroma_minus8 + if (!r.skip_bits(1)) return {0, 0}; // qpprime_y_zero_transform_bypass_flag + std::uint8_t ssmf = 0; + if (!r.read_bit(ssmf)) return {0, 0}; + if (ssmf) { + const int n_lists = (chroma_format_idc != 3) ? 8 : 12; + for (int j = 0; j < n_lists; ++j) { + std::uint8_t present = 0; + if (!r.read_bit(present)) return {0, 0}; + if (present) { + const int sl_size = (j < 6) ? 16 : 64; + int last = 8, next = 8; + for (int k = 0; k < sl_size; ++k) { + if (next != 0) { + std::int32_t delta = 0; + if (!r.read_se(delta)) return {0, 0}; + next = (last + delta + 256) % 256; + } + last = (next == 0) ? last : next; + } + } + } + } + break; + } + + if (!r.skip_ue()) return {0, 0}; // log2_max_frame_num_minus4 + std::uint32_t poc_type = 0; + if (!r.read_ue(poc_type)) return {0, 0}; + if (poc_type == 0) { + if (!r.skip_ue()) return {0, 0}; + } else if (poc_type == 1) { + if (!r.skip_bits(1) || !r.skip_se() || !r.skip_se()) return {0, 0}; + std::uint32_t num_ref = 0; + if (!r.read_ue(num_ref)) return {0, 0}; + for (std::uint32_t k = 0; k < num_ref; ++k) { + if (!r.skip_se()) return {0, 0}; + } + } + if (!r.skip_ue()) return {0, 0}; // max_num_ref_frames + if (!r.skip_bits(1)) return {0, 0}; // gaps_in_frame_num_value_allowed_flag + + std::uint32_t pic_w = 0, pic_h = 0; + if (!r.read_ue(pic_w) || !r.read_ue(pic_h)) return {0, 0}; + std::uint8_t frame_mbs_only = 0; + if (!r.read_bit(frame_mbs_only)) return {0, 0}; + const std::uint32_t width = (pic_w + 1U) * 16U; + std::uint32_t height = (pic_h + 1U) * 16U * (2U - frame_mbs_only); + + if (!frame_mbs_only && !r.skip_bits(1)) return {width, height}; + if (!r.skip_bits(1)) return {width, height}; // direct_8x8_inference_flag + std::uint8_t crop = 0; + if (!r.read_bit(crop)) return {width, height}; + if (crop) { + const std::uint32_t cux = (chroma_format_idc == 0) ? 1U : 2U; + const std::uint32_t cuy = (chroma_format_idc == 0) ? (2U - frame_mbs_only) + : (2U * (2U - frame_mbs_only)); + std::uint32_t cl = 0, cr = 0, ct = 0, cb = 0; + if (!r.read_ue(cl) || !r.read_ue(cr) || !r.read_ue(ct) || !r.read_ue(cb)) { + return {width, height}; + } + return {width - (cl + cr) * cux, height - (ct + cb) * cuy}; + } + return {width, height}; +} + +// Parse {width, height} from a raw HEVC SPS NAL unit (includes 2-byte NAL header). +// Returns {0, 0} on any parse failure. +std::pair +parse_hevc_dimensions(const std::vector& sps_nal) { + if (sps_nal.size() < 16) return {0, 0}; + const std::uint8_t max_sub_layers_minus1 = static_cast((sps_nal[2] >> 1U) & 0x07U); + RbspBitReader r(sps_nal.data() + 2, sps_nal.size() - 2); + if (!r.skip_bits(8)) return {0, 0}; // vps_id(4)+max_sub_layers(3)+temporal_nesting(1) + if (!r.skip_bits(96)) return {0, 0}; // profile_tier_level general_* (12 bytes) + + bool sub_profile[8] = {}, sub_level[8] = {}; + for (int j = 0; j < static_cast(max_sub_layers_minus1); ++j) { + std::uint8_t b = 0; + if (!r.read_bit(b)) return {0, 0}; + sub_profile[j] = (b != 0); + if (!r.read_bit(b)) return {0, 0}; + sub_level[j] = (b != 0); + } + if (max_sub_layers_minus1 < 8 && !r.skip_bits(2U * (8U - max_sub_layers_minus1))) return {0, 0}; + for (int j = 0; j < static_cast(max_sub_layers_minus1); ++j) { + if (sub_profile[j] && !r.skip_bits(88)) return {0, 0}; + if (sub_level[j] && !r.skip_bits(8)) return {0, 0}; + } + + if (!r.skip_ue()) return {0, 0}; // sps_seq_parameter_set_id + std::uint32_t chroma_format_idc = 0; + if (!r.read_ue(chroma_format_idc)) return {0, 0}; + if (chroma_format_idc == 3 && !r.skip_bits(1)) return {0, 0}; + + std::uint32_t width = 0, height = 0; + if (!r.read_ue(width) || !r.read_ue(height)) return {0, 0}; + + std::uint8_t conf_win = 0; + if (!r.read_bit(conf_win)) return {width, height}; + if (conf_win) { + const std::uint32_t swc = (chroma_format_idc == 1 || chroma_format_idc == 2) ? 2U : 1U; + const std::uint32_t shc = (chroma_format_idc == 1) ? 2U : 1U; + std::uint32_t cl = 0, cr = 0, ct = 0, cb = 0; + if (!r.read_ue(cl) || !r.read_ue(cr) || !r.read_ue(ct) || !r.read_ue(cb)) { + return {width, height}; + } + width -= (cl + cr) * swc; + height -= (ct + cb) * shc; + } + return {width, height}; +} + +// Build an hvcC box (HEVCDecoderConfigurationRecord) from VPS, SPS, PPS. +std::vector build_hvcc_box(const HevcParamSets& params) { + if (params.vps.empty() || params.sps.size() < 2 || params.pps.empty()) { + return {}; + } + + // Parse basic info from SPS NAL (after the 2-byte NAL header) + // SPS: nal_header(2) + sps_video_parameter_set_id(4b) + sps_max_sub_layers_minus1(3b) + temporal_id_nesting(1b) + profile_tier_level(...) + const std::uint8_t* sps_data = params.sps.data() + 2; // skip 2-byte NAL header + const std::size_t sps_payload_size = params.sps.size() - 2; + if (sps_payload_size < 13) { + return {}; + } + + // profile_tier_level starts at byte 0 of sps_data after 4-bit vps_id + 3-bit max_sub_layers + 1-bit temporal nesting + // But we need: general_profile_space(2b) + general_tier_flag(1b) + general_profile_idc(5b) = 1 byte + // general_profile_compatibility_flags(32b) = 4 bytes + // constraint_indicator_flags(48b) = 6 bytes + // general_level_idc(8b) = 1 byte + // Total profile_tier_level fixed part = 12 bytes starting after the first byte + const std::uint8_t profile_byte = sps_data[1]; + const std::uint8_t general_profile_space = static_cast((profile_byte >> 6U) & 0x03U); + const std::uint8_t general_tier_flag = static_cast((profile_byte >> 5U) & 0x01U); + const std::uint8_t general_profile_idc = static_cast(profile_byte & 0x1FU); + const std::uint32_t general_profile_compat = (static_cast(sps_data[2]) << 24U) | + (static_cast(sps_data[3]) << 16U) | + (static_cast(sps_data[4]) << 8U) | + static_cast(sps_data[5]); + // constraint_indicator_flags: 6 bytes + std::uint8_t constraint_flags[6]; + for (int j = 0; j < 6; ++j) { + constraint_flags[j] = sps_data[6 + j]; + } + const std::uint8_t general_level_idc = sps_data[12]; + + // Max sub layers from first byte of SPS payload: bits[4:6] = sps_max_sub_layers_minus1 + const std::uint8_t max_sub_layers = static_cast((sps_data[0] >> 1U) & 0x07U); + + // Build HEVCDecoderConfigurationRecord (ISO 14496-15 section 8.3.3.1.2) + std::vector record; + record.push_back(1); // configurationVersion + record.push_back(static_cast((general_profile_space << 6U) | (general_tier_flag << 5U) | general_profile_idc)); + append_be32(record, general_profile_compat); + record.insert(record.end(), constraint_flags, constraint_flags + 6); + record.push_back(general_level_idc); + // min_spatial_segmentation_idc = 0, with reserved bits + append_be16(record, 0xF000U); + // parallelismType = 0 with reserved bits + record.push_back(0xFCU); + // chromaFormat = 1 (4:2:0) with reserved bits + record.push_back(0xFDU); + // bitDepthLumaMinus8 = 0 with reserved bits + record.push_back(0xF8U); + // bitDepthChromaMinus8 = 0 with reserved bits + record.push_back(0xF8U); + // avgFrameRate = 0 + append_be16(record, 0); + // constantFrameRate(2b)=0 + numTemporalLayers(3b) + temporalIdNested(1b) + lengthSizeMinusOne(2b)=3 + record.push_back(static_cast(((max_sub_layers + 1) << 3U) | 0x04U | 0x03U)); + // numOfArrays = 3 (VPS, SPS, PPS) + record.push_back(3); + + // Array entry helper: array_completeness(1b)=1 + reserved(1b)=0 + NAL_unit_type(6b), numNalus, nalUnitLength, nalUnit + auto write_array = [&](std::uint8_t nal_type, const std::vector& nal) { + record.push_back(static_cast(0x80U | nal_type)); // array_completeness=1 + append_be16(record, 1); // numNalus + append_be16(record, static_cast(nal.size())); + record.insert(record.end(), nal.begin(), nal.end()); + }; + + write_array(32, params.vps); // VPS + write_array(33, params.sps); // SPS + write_array(34, params.pps); // PPS + + return make_box("hvcC", record); +} + +// Build an esds box for AAC audio from ADTS header bytes. +// adts_header must be at least 7 bytes (a valid ADTS fixed header). +// Uses expanded (4-byte) descriptor length encoding to match ffmpeg/reference format. +std::vector build_esds_box(std::span adts_header) { + if (adts_header.size() < 7) { + return {}; + } + // Parse ADTS fixed header fields + const std::uint8_t profile = static_cast((adts_header[2] >> 6U) & 0x03U); // 0=Main, 1=LC, 2=SSR, 3=LTP + const std::uint8_t freq_index = static_cast((adts_header[2] >> 2U) & 0x0FU); + const std::uint8_t channel_config = static_cast(((adts_header[2] & 0x01U) << 2U) | + ((adts_header[3] >> 6U) & 0x03U)); + + // Build AudioSpecificConfig (2 bytes for LC-AAC) + const std::uint8_t audio_object_type = static_cast(profile + 1); // AOT = profile + 1 + std::uint8_t asc[2]; + asc[0] = static_cast((audio_object_type << 3U) | (freq_index >> 1U)); + asc[1] = static_cast((freq_index << 7U) | (channel_config << 3U)); + + // Helper: write descriptor tag + expanded 4-byte length (matches ffmpeg/ISO format) + auto write_descr_tag = [](std::vector& out, std::uint8_t tag, std::uint32_t length) { + out.push_back(tag); + out.push_back(static_cast(0x80U | ((length >> 21U) & 0x7FU))); + out.push_back(static_cast(0x80U | ((length >> 14U) & 0x7FU))); + out.push_back(static_cast(0x80U | ((length >> 7U) & 0x7FU))); + out.push_back(static_cast(length & 0x7FU)); + }; + + // Compute descriptor body sizes (without tag+length overhead) + const std::uint32_t dec_specific_body = 2; // AudioSpecificConfig + const std::uint32_t dec_config_body = 13 + 5 + dec_specific_body; // 13 fixed + tag(1)+len(4) + ASC + const std::uint32_t sl_config_body = 1; // predefined + const std::uint32_t es_desc_body = 3 + 5 + dec_config_body + 5 + sl_config_body; // ES_ID(2)+flags(1) + tag+len+DecConfig + tag+len+SLConfig + + // Build the full esds atom content + std::vector esds_content; + esds_content.push_back(0); // version + esds_content.push_back(0); // flags[0] + esds_content.push_back(0); // flags[1] + esds_content.push_back(0); // flags[2] + + // ES_Descriptor (tag=0x03) + write_descr_tag(esds_content, 0x03, es_desc_body); + append_be16(esds_content, 0x0001); // ES_ID = 1 + esds_content.push_back(0x00); // streamDependenceFlag=0 URL_Flag=0 OCRstreamFlag=0 streamPriority=0 + + // DecoderConfigDescriptor (tag=0x04) + write_descr_tag(esds_content, 0x04, dec_config_body); + esds_content.push_back(0x40); // objectTypeIndication = Audio ISO/IEC 14496-3 + esds_content.push_back(0x15); // streamType=5 (audio) upstream=0 reserved=1 + esds_content.push_back(0x00); // bufferSizeDB[0] + esds_content.push_back(0x00); // bufferSizeDB[1] + esds_content.push_back(0x00); // bufferSizeDB[2] + append_be32(esds_content, 128000); // maxBitrate + append_be32(esds_content, 128000); // avgBitrate + + // DecoderSpecificInfo (tag=0x05) + write_descr_tag(esds_content, 0x05, dec_specific_body); + esds_content.push_back(asc[0]); + esds_content.push_back(asc[1]); + + // SLConfigDescriptor (tag=0x06) + write_descr_tag(esds_content, 0x06, sl_config_body); + esds_content.push_back(0x02); // predefined = 2 + + return make_box("esds", esds_content); +} + +// Extract SPS and PPS from AnnexB H.264 keyframe. +// Returns {sps, pps} pair. Either may be empty if not found. +std::pair, std::vector> +extract_h264_sps_pps(std::span payload) { + std::vector sps, pps; + auto nals = extract_nal_units(payload); + for (const auto& nal : nals) { + if (nal.empty()) continue; + const std::uint8_t nal_type = static_cast(nal[0] & 0x1FU); + if (nal_type == 7 && sps.empty()) { // SPS + sps = nal; + } else if (nal_type == 8 && pps.empty()) { // PPS + pps = nal; + } + if (!sps.empty() && !pps.empty()) break; + } + return {sps, pps}; +} + +// Build codec_private bytes for H.264 track from first keyframe's AnnexB data. +std::vector build_h264_codec_private(std::span annexb_keyframe) { + auto [sps, pps] = extract_h264_sps_pps(annexb_keyframe); + return build_avcc_box(sps, pps); +} + +// Build codec_private bytes for AAC from first ADTS frame. +// ADTS sampling frequency table (ISO 14496-3) +constexpr std::uint32_t kAdtsSampleRates[] = { + 96000, 88200, 64000, 48000, 44100, 32000, 24000, 22050, + 16000, 12000, 11025, 8000, 7350 +}; + +// Parse ADTS header to extract sample_rate and channel_count. +// Returns {sample_rate, channel_count} or {0,0} on failure. +std::pair parse_adts_audio_params(std::span adts_header) { + if (adts_header.size() < 7) return {0, 0}; + const std::uint8_t freq_index = static_cast((adts_header[2] >> 2U) & 0x0FU); + const std::uint8_t channel_config = static_cast(((adts_header[2] & 0x01U) << 2U) | + ((adts_header[3] >> 6U) & 0x03U)); + const std::uint32_t sample_rate = freq_index < 13 ? kAdtsSampleRates[freq_index] : 0; + // channel_config 1-7 maps directly to channel count (1=mono, 2=stereo, ..., 6=5.1, 7=7.1) + const std::uint32_t channels = channel_config <= 7 ? channel_config : 0; + return {sample_rate, channels}; +} + +std::vector build_moof_box(std::uint32_t sequence_number, + std::uint32_t track_id, + std::uint64_t base_decode_time, + std::uint32_t sample_duration, + std::uint32_t sample_size, + std::uint32_t sample_flags, + std::int32_t sample_cts_offset, + bool default_base_moof) { + std::vector mfhd_payload; + append_be32(mfhd_payload, sequence_number); + auto mfhd = make_full_box("mfhd", 0, 0, mfhd_payload); + + std::vector tfhd_payload; + append_be32(tfhd_payload, track_id); + std::uint32_t tfhd_flags = default_base_moof ? 0x020000U : 0U; + auto tfhd = make_full_box("tfhd", 0, tfhd_flags, tfhd_payload); + + std::vector tfdt_payload; + append_be64(tfdt_payload, base_decode_time); + auto tfdt = make_full_box("tfdt", 1, 0, tfdt_payload); + + std::vector trun_payload; + append_be32(trun_payload, 1); + append_be32(trun_payload, 0); + append_be32(trun_payload, sample_duration); + append_be32(trun_payload, sample_size); + append_be32(trun_payload, sample_flags); + append_be32(trun_payload, static_cast(sample_cts_offset)); + auto trun = make_full_box("trun", 0, 0x000F01U, trun_payload); + + std::vector traf_payload; + traf_payload.insert(traf_payload.end(), tfhd.begin(), tfhd.end()); + traf_payload.insert(traf_payload.end(), tfdt.begin(), tfdt.end()); + traf_payload.insert(traf_payload.end(), trun.begin(), trun.end()); + auto traf = make_box("traf", traf_payload); + + std::vector moof_payload; + moof_payload.insert(moof_payload.end(), mfhd.begin(), mfhd.end()); + moof_payload.insert(moof_payload.end(), traf.begin(), traf.end()); + auto moof = make_box("moof", moof_payload); + + const std::uint32_t trun_data_offset = static_cast(moof.size() + 8); + const std::size_t trun_offset = moof.size() - trun.size(); + const std::size_t data_offset_field = trun_offset + 8 + 4 + 4; + moof[data_offset_field + 0] = static_cast((trun_data_offset >> 24U) & 0xFFU); + moof[data_offset_field + 1] = static_cast((trun_data_offset >> 16U) & 0xFFU); + moof[data_offset_field + 2] = static_cast((trun_data_offset >> 8U) & 0xFFU); + moof[data_offset_field + 3] = static_cast(trun_data_offset & 0xFFU); + + return moof; +} + +std::vector build_mdat_box(std::span sample_bytes) { + return make_box("mdat", sample_bytes); +} + +class TsPesDemuxer { +public: + explicit TsPesDemuxer(const LiveSrtCallerRuntimeConfig& config) + : config_(config) { + if (config_.has_video_pid) { + video_pid_ = static_cast(config_.video_pid & 0x1FFFU); + } + if (config_.has_audio_pid) { + audio_pid_ = static_cast(config_.audio_pid & 0x1FFFU); + } + } + + void feed(const std::uint8_t* data, + std::size_t size, + const std::function& sample_sink) { + buffer_.insert(buffer_.end(), data, data + static_cast(size)); + + while (buffer_.size() >= 188) { + if (buffer_[0] != 0x47) { + auto sync_it = std::find(buffer_.begin() + 1, buffer_.end(), 0x47); + if (sync_it == buffer_.end()) { + buffer_.clear(); + return; + } + buffer_.erase(buffer_.begin(), sync_it); + continue; + } + + std::array packet{}; + std::copy_n(buffer_.begin(), 188, packet.begin()); + buffer_.erase(buffer_.begin(), buffer_.begin() + 188); + parse_packet(packet, sample_sink); + } + } + +private: + struct PesBuffer { + bool active = false; + bool is_video = false; + std::uint64_t pts90k = 0; + std::vector data; + }; + + void parse_packet(std::span packet, + const std::function& sample_sink) { + const bool payload_unit_start = (packet[1] & 0x40U) != 0; + const std::uint16_t pid = static_cast(((packet[1] & 0x1FU) << 8U) | packet[2]); + const std::uint8_t adaptation_control = static_cast((packet[3] >> 4U) & 0x03U); + if (adaptation_control == 0 || adaptation_control == 2) { + return; + } + + std::size_t offset = 4; + if (adaptation_control == 3) { + const std::size_t adaptation_len = packet[offset]; + offset += 1 + adaptation_len; + if (offset >= packet.size()) { + return; + } + } + + if (pid == 0x0000U) { + parse_pat(packet.subspan(offset), payload_unit_start); + return; + } + if (pmt_pid_.has_value() && pid == *pmt_pid_) { + parse_pmt(packet.subspan(offset), payload_unit_start); + return; + } + if (video_pid_.has_value() && pid == *video_pid_) { + parse_pes(packet.subspan(offset), payload_unit_start, pid, true, sample_sink); + return; + } + if (audio_pid_.has_value() && pid == *audio_pid_) { + parse_pes(packet.subspan(offset), payload_unit_start, pid, false, sample_sink); + } + } + + void parse_pat(std::span payload, bool payload_unit_start) { + if (!payload_unit_start || payload.empty()) { + return; + } + const std::size_t pointer = payload[0]; + if (1 + pointer + 8 > payload.size()) { + return; + } + const std::size_t sec = 1 + pointer; + if (payload[sec] != 0x00) { + return; + } + const std::uint16_t section_length = static_cast(((payload[sec + 1] & 0x0FU) << 8U) | payload[sec + 2]); + const std::size_t end = sec + 3 + section_length; + if (end > payload.size() || section_length < 9) { + return; + } + for (std::size_t cursor = sec + 8; cursor + 4 <= end - 4; cursor += 4) { + const std::uint16_t program_number = read_be16(payload, cursor); + const std::uint16_t pid = static_cast(((payload[cursor + 2] & 0x1FU) << 8U) | payload[cursor + 3]); + if (program_number == 0) { + continue; + } + if (!config_.has_program_number || program_number == config_.program_number) { + pmt_pid_ = pid; + return; + } + } + } + + void parse_pmt(std::span payload, bool payload_unit_start) { + if (!payload_unit_start || payload.empty()) { + return; + } + const std::size_t pointer = payload[0]; + if (1 + pointer + 12 > payload.size()) { + return; + } + const std::size_t sec = 1 + pointer; + if (payload[sec] != 0x02) { + return; + } + const std::uint16_t section_length = static_cast(((payload[sec + 1] & 0x0FU) << 8U) | payload[sec + 2]); + const std::size_t end = sec + 3 + section_length; + if (end > payload.size() || section_length < 13) { + return; + } + + const std::uint16_t program_info_length = static_cast(((payload[sec + 10] & 0x0FU) << 8U) | payload[sec + 11]); + std::size_t cursor = sec + 12 + program_info_length; + while (cursor + 5 <= end - 4) { + const std::uint8_t stream_type = payload[cursor]; + const std::uint16_t elementary_pid = + static_cast(((payload[cursor + 1] & 0x1FU) << 8U) | payload[cursor + 2]); + const std::uint16_t es_info_len = + static_cast(((payload[cursor + 3] & 0x0FU) << 8U) | payload[cursor + 4]); + + if (!video_pid_.has_value() && (stream_type == 0x1B || stream_type == 0x24 || stream_type == 0x02)) { + video_pid_ = elementary_pid; + video_stream_type_ = stream_type; + } + if (!audio_pid_.has_value() && (stream_type == 0x0F || stream_type == 0x11 || stream_type == 0x03 || stream_type == 0x04)) { + audio_pid_ = elementary_pid; + audio_stream_type_ = stream_type; + } + cursor += 5 + es_info_len; + } + pmt_parsed_ = true; + } + + void parse_pes(std::span payload, + bool payload_unit_start, + std::uint16_t pid, + bool is_video, + const std::function& sample_sink) { + auto& pes = pes_by_pid_[pid]; + if (payload_unit_start) { + flush_pes(pes, sample_sink); + pes = PesBuffer{}; + pes.active = true; + pes.is_video = is_video; + if (payload.size() < 9) { + return; + } + if (!(payload[0] == 0x00 && payload[1] == 0x00 && payload[2] == 0x01)) { + return; + } + const std::uint8_t flags = payload[7]; + const std::uint8_t header_len = payload[8]; + std::size_t cursor = 9; + if ((flags & 0x80U) != 0 && cursor + 5 <= payload.size()) { + pes.pts90k = static_cast(((payload[cursor] >> 1U) & 0x07U)) << 30U; + pes.pts90k |= static_cast(payload[cursor + 1]) << 22U; + pes.pts90k |= static_cast((payload[cursor + 2] >> 1U) & 0x7FU) << 15U; + pes.pts90k |= static_cast(payload[cursor + 3]) << 7U; + pes.pts90k |= static_cast((payload[cursor + 4] >> 1U) & 0x7FU); + } + cursor = 9 + header_len; + if (cursor < payload.size()) { + pes.data.insert(pes.data.end(), payload.begin() + static_cast(cursor), payload.end()); + } + return; + } + + if (!pes.active) { + return; + } + pes.data.insert(pes.data.end(), payload.begin(), payload.end()); + } + + void flush_pes(PesBuffer& pes, const std::function& sample_sink) { + if (!pes.active || pes.data.empty()) { + return; + } + const std::uint8_t stream_type = pes.is_video ? video_stream_type_ : audio_stream_type_; + if (pes.is_video) { + EsSample sample; + sample.is_video = true; + sample.pts90k = pes.pts90k; + sample.stream_type = stream_type; + sample.payload = std::move(pes.data); + const VideoCodec codec = detect_video_codec_from_stream_type(stream_type); + if (codec == VideoCodec::kHevc) { + sample.keyframe = hevc_annexb_has_irap(sample.payload); + } else { + sample.keyframe = h264_annexb_has_idr(sample.payload); + } + sample_sink(std::move(sample)); + } else { + // Split ADTS audio into individual AAC frames. + // Each frame becomes its own EsSample with interpolated PTS. + const bool is_adts = (stream_type == 0x0F || stream_type == 0x11) && + pes.data.size() >= 7 && + pes.data[0] == 0xFF && + (pes.data[1] & 0xF0U) == 0xF0U; + if (is_adts) { + std::size_t offset = 0; + std::size_t frame_index = 0; + // Detect sample rate from first ADTS header if not yet known + if (audio_sample_rate_ == 0) { + auto [rate, channels] = parse_adts_audio_params( + std::span(pes.data.data(), std::min(pes.data.size(), std::size_t{9}))); + if (rate > 0) { + audio_sample_rate_ = rate; + } + } + // AAC frame PTS: use rational per-frame calculation to avoid + // truncation drift (important for 44.1kHz where 1024*90000/44100 + // is not an integer). + const std::uint64_t aac_sample_rate = + audio_sample_rate_ > 0 ? static_cast(audio_sample_rate_) : 48000ULL; + while (offset + 7 <= pes.data.size()) { + if (pes.data[offset] != 0xFF || (pes.data[offset + 1] & 0xF0U) != 0xF0U) { + break; + } + const bool protection_absent = (pes.data[offset + 1] & 0x01U) != 0; + const std::size_t header_len = protection_absent ? 7 : 9; + if (offset + header_len > pes.data.size()) { + break; + } + const std::uint16_t frame_length = + static_cast(((pes.data[offset + 3] & 0x03U) << 11U) | + (pes.data[offset + 4] << 3U) | + ((pes.data[offset + 5] >> 5U) & 0x07U)); + if (frame_length < header_len || offset + frame_length > pes.data.size()) { + break; + } + EsSample sample; + sample.is_video = false; + // Rational PTS: (frame_index * 1024 * 90000 + rate/2) / rate + sample.pts90k = pes.pts90k + + (frame_index * 1024ULL * 90000ULL + aac_sample_rate / 2) / aac_sample_rate; + sample.stream_type = stream_type; + sample.keyframe = false; + // Preserve ADTS header on first frame for codec discovery + if (frame_index == 0) { + const std::size_t copy_len = std::min(header_len, std::size_t{9}); + std::copy_n(pes.data.begin() + static_cast(offset), + copy_len, sample.adts_header.begin()); + sample.adts_header_len = static_cast(copy_len); + } + // Store raw AAC frame data (without ADTS header) + const std::size_t aac_len = frame_length - header_len; + sample.payload.assign( + pes.data.begin() + static_cast(offset + header_len), + pes.data.begin() + static_cast(offset + header_len + aac_len)); + sample_sink(std::move(sample)); + offset += frame_length; + ++frame_index; + } + // If we couldn't parse any ADTS frames, emit raw + if (frame_index == 0) { + EsSample sample; + sample.is_video = false; + sample.pts90k = pes.pts90k; + sample.stream_type = stream_type; + sample.keyframe = false; + sample.payload = std::move(pes.data); + sample_sink(std::move(sample)); + } + } else { + // Non-ADTS audio: emit as single sample + EsSample sample; + sample.is_video = false; + sample.pts90k = pes.pts90k; + sample.stream_type = stream_type; + sample.keyframe = false; + sample.payload = std::move(pes.data); + sample_sink(std::move(sample)); + } + } + pes = PesBuffer{}; + } + + LiveSrtCallerRuntimeConfig config_; + std::vector buffer_; + std::optional pmt_pid_; + std::optional video_pid_; + std::optional audio_pid_; + std::uint8_t video_stream_type_ = 0; + std::uint8_t audio_stream_type_ = 0; + std::uint32_t audio_sample_rate_ = 0; + bool pmt_parsed_ = false; + std::map pes_by_pid_; + +public: + bool pmt_parsed() const { return pmt_parsed_; } + bool has_audio() const { return audio_pid_.has_value(); } + void set_audio_sample_rate(std::uint32_t rate) { audio_sample_rate_ = rate; } +}; + +std::uint64_t to_us_from_90k(std::uint64_t pts90k) { + return (pts90k * 1000000ULL) / 90000ULL; +} + +MediaFragment build_fragment_from_sample(const LiveSrtCallerRuntimeConfig& config, + CallerTrackState& state, + EsSample&& sample) { + const std::string track_name = sample.is_video ? state.video_track_name : state.audio_track_name; + const std::uint32_t track_id = sample.is_video ? state.video_track_id : state.audio_track_id; + + if (sample.is_video && config.fragment_on_keyframe && sample.keyframe) { + if (state.first_video_keyframe_seen) { + ++state.group_id; + } + state.first_video_keyframe_seen = true; + state.object_id_by_track.clear(); + } + + if (config.fragment_on_keyframe && !state.first_video_keyframe_seen) { + return MediaFragment{}; + } + + const std::uint64_t pts_us = to_us_from_90k(sample.pts90k); + const std::uint64_t last_pts = state.last_pts_by_track[track_name]; + std::uint64_t duration_us = 0; + if (last_pts != 0 && pts_us > last_pts) { + duration_us = pts_us - last_pts; + // Clamp unreasonably large durations (>500ms) — likely a PTS discontinuity + if (duration_us > 500000) { + duration_us = state.last_duration_us_by_track[track_name]; + } + } + if (duration_us == 0) { + // Use last known good duration, or estimate from frame rate + duration_us = state.last_duration_us_by_track[track_name]; + if (duration_us == 0) { + duration_us = sample.is_video ? 33333 : ((1024ULL * 1000000ULL) / (state.audio_timescale > 0 ? state.audio_timescale : 48000U)); + } + } + state.last_pts_by_track[track_name] = pts_us; + state.last_duration_us_by_track[track_name] = duration_us; + + if (sample.is_video && state.video_codec == VideoCodec::kUnknown) { + state.video_codec = detect_video_codec_from_stream_type(sample.stream_type); + if (state.video_codec == VideoCodec::kUnknown) { + state.video_codec = detect_video_codec_from_annexb(sample.payload); + } + if (state.video_codec == VideoCodec::kUnknown) { + state.video_codec = VideoCodec::kH264; + } + } + + std::vector sample_bytes; + if (sample.is_video) { + sample_bytes = annexb_to_avcc(sample.payload); + } else { + // Audio payload is already raw AAC (ADTS stripped in flush_pes) + sample_bytes = std::move(sample.payload); + } + + // Accumulation-based decode_time: monotonically increasing, never resets. + // MSE requires timestamps that never go backwards. PTS-based computation + // can produce non-monotonic values due to SRT jitter, causing the player + // to drop "overlapping" payloads. + const std::uint32_t timescale = sample.is_video ? state.video_timescale : state.audio_timescale; + + // AAC audio: always use exactly 1024 samples per frame in the audio timescale. + // Deriving from duration_us can produce 1023 or 1025 due to rounding. + const std::uint32_t sample_duration = !sample.is_video + ? 1024U + : static_cast((duration_us * static_cast(timescale)) / 1000000ULL); + + const std::uint64_t base_decode_time = state.decode_time_by_track[track_name]; + state.decode_time_by_track[track_name] += (sample_duration == 0 ? 1U : sample_duration); + const std::uint32_t sample_flags = (sample.is_video && !sample.keyframe) ? 0x00010000U : 0x02000000U; + + auto moof = build_moof_box(state.moof_sequence++, + track_id, + base_decode_time, + sample_duration == 0 ? 1 : sample_duration, + static_cast(sample_bytes.size()), + sample_flags, + 0, + config.default_base_moof); + auto mdat = build_mdat_box(sample_bytes); + + MediaFragment fragment; + fragment.group_id = state.group_id; + fragment.object_id = state.object_id_by_track[track_name]++; + fragment.track_name = track_name; + fragment.start_time_us = pts_us; + fragment.duration_us = duration_us; + fragment.earliest_presentation_time_us = pts_us; + fragment.sap_type = sample.is_video ? (sample.keyframe ? 1 : 2) : 1; + fragment.is_video_keyframe = sample.is_video && sample.keyframe; + fragment.creation_time_us = static_cast( + std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()).count()); + fragment.payload.owned_bytes.reserve(moof.size() + mdat.size()); + fragment.payload.owned_bytes.insert(fragment.payload.owned_bytes.end(), moof.begin(), moof.end()); + fragment.payload.owned_bytes.insert(fragment.payload.owned_bytes.end(), mdat.begin(), mdat.end()); + fragment.payload.span = ByteSpan{.offset = 0, .size = fragment.payload.owned_bytes.size()}; + return fragment; +} + +TrackDescription make_track(std::uint32_t track_id, + std::string track_name, + bool is_video, + VideoCodec video_codec = VideoCodec::kH264) { + TrackDescription track; + track.track_id = track_id; + track.handler_type = is_video ? "vide" : "soun"; + if (is_video) { + if (video_codec == VideoCodec::kHevc) { + track.codec = "hvc1.1.6.L120.B0"; + track.sample_entry_type = "hvc1"; + } else { + track.codec = "avc1.42E01E"; + track.sample_entry_type = "avc1"; + } + } else { + track.codec = "mp4a.40.2"; + track.sample_entry_type = "mp4a"; + } + track.track_name = std::move(track_name); + track.packaging = "cmaf"; + track.mime_type = is_video ? "video/mp4" : "audio/mp4"; + track.timescale = is_video ? 90000 : 48000; + track.width = 0; + track.height = 0; + track.channel_count = is_video ? 0 : 2; + track.sample_rate = is_video ? 0 : 48000; + track.frame_rate = is_video ? 30.0 : 0.0; + return track; +} + +std::pair split_host_port(std::string_view endpoint) { + const auto pos = endpoint.rfind(':'); + if (pos == std::string_view::npos || pos == 0 || pos + 1 >= endpoint.size()) { + throw std::runtime_error("invalid SRT endpoint, expected host:port"); + } + const std::string host(endpoint.substr(0, pos)); + const int port = std::stoi(std::string(endpoint.substr(pos + 1))); + if (port <= 0 || port > 65535) { + throw std::runtime_error("invalid SRT endpoint port"); + } + return {host, static_cast(port)}; +} + +} // namespace + +LiveSrtIngestManager::LiveSrtIngestManager(std::vector callers, + FragmentSink sink, + std::atomic& stop_requested) + : callers_(std::move(callers)), + sink_(std::move(sink)), + stop_requested_(stop_requested) { + std::uint32_t next_track_id = 1000; + for (const auto& caller : callers_) { + bootstrap_.tracks.push_back(make_track(next_track_id++, caller.id + "_video", true, VideoCodec::kH264)); + bootstrap_.tracks.push_back(make_track(next_track_id++, caller.id + "_audio", false)); + } + bootstrap_.init_segment = build_init_segment_from_tracks(bootstrap_.tracks); + std::cout << "[SRT] Ingest manager created with " << callers_.size() << " caller(s)\n"; +} + +transport::TransportStatus LiveSrtIngestManager::start() { +#if !defined(OPENMOQ_HAS_SRT) + return transport::TransportStatus::failure("SRT ingest requested but this build does not include libsrt"); +#else + if (callers_.empty()) { + return transport::TransportStatus::failure("SRT ingest requested with no callers configured"); + } + + if (srt_startup() != 0) { + return transport::TransportStatus::failure("srt_startup failed"); + } + + struct CodecDiscovery { + std::mutex mutex; + std::condition_variable cv; + std::size_t discovered = 0; + std::vector ready; + std::vector video_private_ready; + std::vector audio_private_ready; + std::atomic phase_done{false}; + + explicit CodecDiscovery(std::size_t count) + : ready(count, false), video_private_ready(count, false), audio_private_ready(count, false) {} + + bool all_codec_private_ready() const { + for (std::size_t j = 0; j < video_private_ready.size(); ++j) { + if (!video_private_ready[j] || !audio_private_ready[j]) return false; + } + return true; + } + }; + auto discovery = std::make_shared(callers_.size()); + + for (std::size_t i = 0; i < callers_.size(); ++i) { + const auto caller = callers_[i]; + const auto video_track = bootstrap_.tracks[i * 2]; + const auto audio_track = bootstrap_.tracks[i * 2 + 1]; + worker_threads_.emplace_back([this, + i, + caller, + video_track, + audio_track, + discovery]() { + try { + auto [host, port] = split_host_port(caller.endpoint); + CallerTrackState state; + state.video_track_name = video_track.track_name; + state.audio_track_name = audio_track.track_name; + state.video_track_id = video_track.track_id; + state.audio_track_id = audio_track.track_id; + state.video_timescale = video_track.timescale == 0 ? 90000 : video_track.timescale; + state.audio_timescale = audio_track.timescale == 0 ? 48000 : audio_track.timescale; + + SRTSOCKET sock = srt_create_socket(); + if (sock == SRT_INVALID_SOCK) { + return; + } + + const int latency = static_cast(caller.latency_ms); + srt_setsockopt(sock, 0, SRTO_LATENCY, &latency, sizeof(latency)); + + // Set receive timeout so srt_recv returns periodically, allowing + // the thread to check stop_requested_ and exit cleanly. + const int rcv_timeout_ms = 200; + srt_setsockopt(sock, 0, SRTO_RCVTIMEO, &rcv_timeout_ms, sizeof(rcv_timeout_ms)); + + struct addrinfo hints {}; + hints.ai_family = AF_UNSPEC; + hints.ai_socktype = SOCK_DGRAM; + struct addrinfo* result = nullptr; + if (getaddrinfo(host.c_str(), std::to_string(port).c_str(), &hints, &result) != 0 || result == nullptr) { + srt_close(sock); + return; + } + + const int connect_rc = srt_connect(sock, result->ai_addr, static_cast(result->ai_addrlen)); + freeaddrinfo(result); + if (connect_rc == SRT_ERROR) { + std::cerr << "[SRT] Connection FAILED to " << caller.endpoint << "\n"; + srt_close(sock); + return; + } + std::cout << "[SRT] Connected to " << caller.endpoint + << " (latency=" << caller.latency_ms << "ms)\n"; + + TsPesDemuxer demuxer(caller); + std::array recv_buf{}; + bool video_codec_private_captured = false; + bool audio_codec_private_captured = false; + while (!stop_requested_.load()) { + const int received = srt_recv(sock, reinterpret_cast(recv_buf.data()), static_cast(recv_buf.size())); + if (received <= 0) { + // On any recv failure, check socket state to distinguish + // a transient timeout from a dead connection. + const SRT_SOCKSTATUS sock_state = srt_getsockstate(sock); + if (sock_state == SRTS_BROKEN || sock_state == SRTS_CLOSED || + sock_state == SRTS_NONEXIST) { + std::cerr << "[SRT] Connection lost to " << caller.endpoint << "\n"; + break; + } + // Timeout or transient error — loop back to check stop flag. + continue; + } + demuxer.feed(recv_buf.data(), static_cast(received), + [&](EsSample&& sample) { + if (sample.is_video && state.video_codec == VideoCodec::kUnknown) { + state.video_codec = detect_video_codec_from_stream_type(sample.stream_type); + if (state.video_codec == VideoCodec::kUnknown) { + state.video_codec = detect_video_codec_from_annexb(sample.payload); + } + if (state.video_codec != VideoCodec::kUnknown && + !discovery->phase_done.load(std::memory_order_acquire)) { + std::lock_guard lock(discovery->mutex); + if (discovery->phase_done.load(std::memory_order_relaxed)) { + // Main thread finalized; skip track mutation. + } else if (!discovery->ready[i]) { + discovery->ready[i] = true; + ++discovery->discovered; + const std::size_t video_index = i * 2; + if (video_index < bootstrap_.tracks.size()) { + bootstrap_.tracks[video_index] = make_track( + bootstrap_.tracks[video_index].track_id, + bootstrap_.tracks[video_index].track_name, + true, + state.video_codec); + } + } + discovery->cv.notify_one(); + } + } + // Extract video codec_private (avcC/hvcC) from first keyframe + if (sample.is_video && !video_codec_private_captured && + sample.keyframe && + !discovery->phase_done.load(std::memory_order_acquire)) { + std::vector codec_priv; + std::string codec_str_update; + std::pair dim_update{0, 0}; + if (state.video_codec == VideoCodec::kHevc) { + auto params = extract_hevc_param_sets(sample.payload); + codec_priv = build_hvcc_box(params); + dim_update = parse_hevc_dimensions(params.sps); + if (!codec_priv.empty() && params.sps.size() > 12) { + // Build codec string: hvc1... + const std::uint8_t* s = params.sps.data() + 2; + const std::uint8_t prof_idc = static_cast(s[1] & 0x1FU); + const std::uint8_t tier_flag = static_cast((s[1] >> 5U) & 0x01U); + const std::uint32_t compat = (static_cast(s[2]) << 24U) | + (static_cast(s[3]) << 16U) | + (static_cast(s[4]) << 8U) | + static_cast(s[5]); + const std::uint8_t level_idc = s[12]; + char buf[64]; + std::snprintf(buf, sizeof(buf), "hvc1.%u.%X.%c%u", + prof_idc, compat, tier_flag ? 'H' : 'L', level_idc / 3); + codec_str_update = buf; + } + } else { + auto [sps, pps] = extract_h264_sps_pps(sample.payload); + codec_priv = build_avcc_box(sps, pps); + dim_update = parse_h264_dimensions(sps); + if (!codec_priv.empty() && sps.size() >= 4) { + char buf[32]; + std::snprintf(buf, sizeof(buf), "avc1.%02X%02X%02X", + sps[1], sps[2], sps[3]); + codec_str_update = buf; + } + } + if (!codec_priv.empty()) { + video_codec_private_captured = true; + std::lock_guard lock(discovery->mutex); + if (!discovery->phase_done.load(std::memory_order_relaxed)) { + const std::size_t video_index = i * 2; + if (video_index < bootstrap_.tracks.size()) { + bootstrap_.tracks[video_index].codec_private = std::move(codec_priv); + if (!codec_str_update.empty()) { + bootstrap_.tracks[video_index].codec = codec_str_update; + } + if (dim_update.first > 0 && dim_update.second > 0) { + bootstrap_.tracks[video_index].width = dim_update.first; + bootstrap_.tracks[video_index].height = dim_update.second; + } + discovery->video_private_ready[i] = true; + std::cout << "[SRT] Video " + << (state.video_codec == VideoCodec::kHevc ? "hvcC" : "avcC") + << " extracted (" + << bootstrap_.tracks[video_index].codec_private.size() + << " bytes) codec=" << bootstrap_.tracks[video_index].codec + << " from " << caller.id << "\n"; + } + } + discovery->cv.notify_one(); + } + } + // Extract audio codec_private (esds) from first ADTS frame + if (!sample.is_video && !audio_codec_private_captured && + !discovery->phase_done.load(std::memory_order_acquire) && + sample.adts_header_len >= 7) { + auto codec_priv = build_esds_box( + std::span(sample.adts_header.data(), sample.adts_header_len)); + if (!codec_priv.empty()) { + audio_codec_private_captured = true; + // Parse actual sample_rate and channel_count from ADTS header + auto [adts_rate, adts_channels] = parse_adts_audio_params( + std::span(sample.adts_header.data(), sample.adts_header_len)); + if (adts_rate > 0) { + state.audio_timescale = adts_rate; + demuxer.set_audio_sample_rate(adts_rate); + } + std::lock_guard lock(discovery->mutex); + if (!discovery->phase_done.load(std::memory_order_relaxed)) { + const std::size_t audio_index = i * 2 + 1; + if (audio_index < bootstrap_.tracks.size()) { + bootstrap_.tracks[audio_index].codec_private = std::move(codec_priv); + if (adts_rate > 0) { + bootstrap_.tracks[audio_index].sample_rate = adts_rate; + bootstrap_.tracks[audio_index].timescale = adts_rate; + } + if (adts_channels > 0) { + bootstrap_.tracks[audio_index].channel_count = adts_channels; + } + discovery->audio_private_ready[i] = true; + std::cout << "[SRT] Audio esds extracted (" + << bootstrap_.tracks[audio_index].codec_private.size() + << " bytes) from " << caller.id << "\n"; + } + } + discovery->cv.notify_one(); + } + } + auto fragment = build_fragment_from_sample(caller, state, std::move(sample)); + if (fragment.payload.owned_bytes.empty()) { + return; + } + sink_(std::move(fragment)); + }); + // After feed: if PMT parsed and no audio PID found, mark audio discovery done early + if (!audio_codec_private_captured && demuxer.pmt_parsed() && !demuxer.has_audio() && + !discovery->phase_done.load(std::memory_order_acquire)) { + audio_codec_private_captured = true; + std::lock_guard lock(discovery->mutex); + if (!discovery->phase_done.load(std::memory_order_relaxed)) { + discovery->audio_private_ready[i] = true; + std::cout << "[SRT] No audio PID in PMT for " << caller.id << "\n"; + } + discovery->cv.notify_one(); + } + } + if (!video_codec_private_captured) { + std::cerr << "[SRT] Worker exited without discovering video codec_private for " + << caller.id << "\n"; + } + srt_close(sock); + } catch (...) { + } + }); + } + + { + std::unique_lock lock(discovery->mutex); + discovery->cv.wait_for(lock, + std::chrono::seconds(5), + [&]() { + return discovery->discovered >= callers_.size() && + discovery->all_codec_private_ready(); + }); + // Set phase_done while holding the mutex so workers that already passed + // the phase_done check but haven't acquired the mutex yet will re-check. + discovery->phase_done.store(true, std::memory_order_release); + + // Remove tracks that have no codec_private (e.g. audio track when feed has no audio). + // This ensures the catalog only advertises tracks that are actually present in the feed. + bootstrap_.tracks.erase( + std::remove_if(bootstrap_.tracks.begin(), bootstrap_.tracks.end(), + [](const TrackDescription& t) { + return t.codec_private.empty(); + }), + bootstrap_.tracks.end()); + + bootstrap_.init_segment = build_init_segment_from_tracks(bootstrap_.tracks); + } + std::cout << "[SRT] Codec discovery complete. Init segment rebuilt (" + << bootstrap_.init_segment.size() << " bytes)\n"; + for (const auto& track : bootstrap_.tracks) { + std::cout << "[SRT] Track: " << track.track_name + << " codec=" << track.codec + << " codec_private=" << track.codec_private.size() << " bytes\n"; + } + + return transport::TransportStatus::success(); +#endif +} + +void LiveSrtIngestManager::join() { + for (auto& worker : worker_threads_) { + if (worker.joinable()) { + worker.join(); + } + } +#if defined(OPENMOQ_HAS_SRT) + srt_cleanup(); +#endif +} + +const LiveSrtBootstrap& LiveSrtIngestManager::bootstrap() const { + return bootstrap_; +} + +std::vector LiveSrtIngestManager::build_synthetic_init_segment(const std::vector& tracks) { + return build_init_segment_from_tracks(tracks); +} + +} // namespace openmoq::publisher diff --git a/src/main.cpp b/src/main.cpp index 2b486a2..16f3815 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -1,11 +1,13 @@ #include "openmoq/publisher/cmaf_segmenter.h" #include "openmoq/publisher/cli_options.h" #include "openmoq/publisher/cmsf_packager.h" +#include "openmoq/publisher/live_srt_config.h" #include "openmoq/publisher/publisher_api.h" #include "openmoq/publisher/transport/moqt_session.h" #include #include +#include int main(int argc, char** argv) { using namespace openmoq::publisher; @@ -13,10 +15,13 @@ int main(int argc, char** argv) { try { const CliOptions options = parse_cli_options(argc, argv); - // Live stdin mode: when reading from stdin with an endpoint, use - // incremental streaming instead of buffering everything to EOF. - const bool live_stdin = options.input_source.kind == InputSourceKind::kStdin - && options.endpoint.has_value(); + const bool stdin_source_available = options.input_source.kind == InputSourceKind::kStdin; + const bool live_srt = options.live_source == LiveSourceKind::kSrt && + options.srt_config_path.has_value() && options.endpoint.has_value(); + const bool live_stdin = !live_srt && + (options.live_source == LiveSourceKind::kAuto || + options.live_source == LiveSourceKind::kStdin) && + stdin_source_available && options.endpoint.has_value(); const PublisherConfig config{ .draft_version = options.draft_version, .track_namespace = options.track_namespace, @@ -31,9 +36,37 @@ int main(int argc, char** argv) { }; Publisher publisher(config); - if (live_stdin) { + if (live_stdin || live_srt) { + LiveIngestConfig ingest; + ingest.use_stdin = live_stdin; + + if (live_srt) { + std::cerr << "live_source=srt" << std::endl; + const LiveSrtConfig srt_config = parse_live_srt_config_file(*options.srt_config_path); + for (const auto& caller : srt_config.srt_callers) { + LiveSrtCaller live_caller; + live_caller.id = caller.id; + live_caller.endpoint = caller.srt.host + ":" + std::to_string(caller.srt.port); + live_caller.fragment_on_keyframe = caller.cmaf.fragment_on_keyframe; + live_caller.empty_moov = caller.cmaf.empty_moov; + live_caller.default_base_moof = caller.cmaf.default_base_moof; + live_caller.separate_moof_per_track = caller.cmaf.separate_moof_per_track; + live_caller.target_fragment_duration_ms = caller.cmaf.target_fragment_duration_ms; + live_caller.latency_ms = caller.srt.latency_ms; + live_caller.auto_detect_program = caller.mpegts.auto_detect_program; + live_caller.program_number = caller.mpegts.program_number; + live_caller.video_pid = caller.mpegts.video_pid; + live_caller.audio_pid = caller.mpegts.audio_pid; + ingest.srt_callers.push_back(std::move(live_caller)); + } + } else { + std::cerr << "live_source=stdin" << std::endl; + } + + std::istream* stdin_input = live_stdin ? &std::cin : nullptr; const auto status = publisher.publish_live( - std::cin, + ingest, + stdin_input, *options.endpoint, options.tls, options.endpoint_alpn_overridden); diff --git a/src/publisher_api.cpp b/src/publisher_api.cpp index 478d9bc..ffd04d8 100644 --- a/src/publisher_api.cpp +++ b/src/publisher_api.cpp @@ -194,6 +194,16 @@ transport::TransportStatus Publisher::publish_live(std::istream& input, const transport::EndpointConfig& endpoint, const transport::TlsConfig& tls, bool endpoint_alpn_overridden) const { + LiveIngestConfig ingest; + ingest.use_stdin = true; + return publish_live(ingest, &input, endpoint, tls, endpoint_alpn_overridden); +} + +transport::TransportStatus Publisher::publish_live(const LiveIngestConfig& ingest, + std::istream* stdin_input, + const transport::EndpointConfig& endpoint, + const transport::TlsConfig& tls, + bool endpoint_alpn_overridden) const { if (!transport_factory_) { return transport::TransportStatus::failure("publisher transport factory is not configured"); } @@ -221,7 +231,30 @@ transport::TransportStatus Publisher::publish_live(std::istream& input, return transport::TransportStatus::failure(error); } - status = active->session->publish_live(input, config_.draft_version, config_.split_cmaf_chunks); + transport::LiveIngestOptions session_ingest; + session_ingest.use_stdin = ingest.use_stdin; + session_ingest.srt_callers.reserve(ingest.srt_callers.size()); + for (const auto& caller : ingest.srt_callers) { + transport::LiveSrtCallerOptions session_caller; + session_caller.id = caller.id; + session_caller.endpoint = caller.endpoint; + session_caller.fragment_on_keyframe = caller.fragment_on_keyframe; + session_caller.empty_moov = caller.empty_moov; + session_caller.default_base_moof = caller.default_base_moof; + session_caller.separate_moof_per_track = caller.separate_moof_per_track; + session_caller.target_fragment_duration_ms = caller.target_fragment_duration_ms; + session_caller.latency_ms = caller.latency_ms; + session_caller.auto_detect_program = caller.auto_detect_program; + session_caller.program_number = caller.program_number; + session_caller.video_pid = caller.video_pid; + session_caller.audio_pid = caller.audio_pid; + session_ingest.srt_callers.push_back(std::move(session_caller)); + } + + status = active->session->publish_live(session_ingest, + stdin_input, + config_.draft_version, + config_.split_cmaf_chunks); if (!status.ok) { const std::string error = "transport live publish failed: " + status.message; static_cast(active->session->close(0)); diff --git a/src/transport/moqt_session.cpp b/src/transport/moqt_session.cpp index 9955615..97131b6 100644 --- a/src/transport/moqt_session.cpp +++ b/src/transport/moqt_session.cpp @@ -1,9 +1,11 @@ #include "openmoq/publisher/transport/moqt_session.h" #include "openmoq/publisher/transport/moqt_control_messages.h" #include "openmoq/publisher/cmaf_segmenter.h" +#include "openmoq/publisher/live_srt_ingest.h" #include "openmoq/publisher/mp4_box.h" #include +#include #include #include #include @@ -2886,6 +2888,356 @@ TransportStatus MoqtSession::publish(const openmoq::publisher::PublishPlan& plan peer_max_request_id_); } +TransportStatus MoqtSession::publish_live(const LiveIngestOptions& ingest, + std::istream* stdin_input, + openmoq::publisher::DraftVersion draft_version, + bool split_cmaf_chunks) { + if (ingest.use_stdin && !ingest.srt_callers.empty()) { + return TransportStatus::failure("mixed stdin+SRT ingest is not supported; use either stdin or srt"); + } + if (ingest.use_stdin && stdin_input == nullptr) { + return TransportStatus::failure("live ingest requested stdin but no stdin stream was provided"); + } + if (!ingest.use_stdin && ingest.srt_callers.empty()) { + return TransportStatus::failure("live ingest requires at least one active source"); + } + if (ingest.use_stdin) { + return publish_live(*stdin_input, draft_version, split_cmaf_chunks); + } + + // SRT-only path below. + if (transport_.state() != ConnectionState::kConnected) { + return TransportStatus::failure("transport is not connected"); + } + + TransportStatus status = ensure_setup(draft_version); + if (!status.ok) { + return status; + } + std::cout << "connection_id=" << transport_.connection_id() << '\n' << std::flush; + + struct LiveMediaQueue { + std::mutex mutex; + std::deque fragments; + bool eof = false; + }; + auto queue = std::make_shared(); + std::atomic stop_requested = false; + + std::vector srt_callers; + srt_callers.reserve(ingest.srt_callers.size()); + for (const auto& caller : ingest.srt_callers) { + openmoq::publisher::LiveSrtCallerRuntimeConfig config; + config.id = caller.id; + config.endpoint = caller.endpoint; + config.fragment_on_keyframe = caller.fragment_on_keyframe; + config.empty_moov = caller.empty_moov; + config.default_base_moof = caller.default_base_moof; + config.separate_moof_per_track = caller.separate_moof_per_track; + config.target_fragment_duration_ms = caller.target_fragment_duration_ms; + config.latency_ms = caller.latency_ms; + config.auto_detect_program = caller.auto_detect_program; + config.program_number = caller.program_number.value_or(0); + config.has_program_number = caller.program_number.has_value(); + config.video_pid = caller.video_pid.value_or(0); + config.has_video_pid = caller.video_pid.has_value(); + config.audio_pid = caller.audio_pid.value_or(0); + config.has_audio_pid = caller.audio_pid.has_value(); + srt_callers.push_back(std::move(config)); + } + + openmoq::publisher::LiveSrtIngestManager srt_manager( + std::move(srt_callers), + [queue](openmoq::publisher::MediaFragment&& fragment) { + std::lock_guard lock(queue->mutex); + queue->fragments.push_back(std::move(fragment)); + }, + stop_requested); + + status = srt_manager.start(); + if (!status.ok) { + return status; + } + const auto& srt_bootstrap = srt_manager.bootstrap(); + const auto& tracks = srt_bootstrap.tracks; + + if (tracks.empty()) { + stop_requested = true; + srt_manager.join(); + return TransportStatus::failure("no live tracks available from SRT source"); + } + + const std::vector synthetic_init = + openmoq::publisher::LiveSrtIngestManager::build_synthetic_init_segment(tracks); + openmoq::publisher::LiveCatalog live_catalog = + openmoq::publisher::build_live_catalog(tracks, synthetic_init, true); + + std::thread srt_join_thread([&srt_manager, queue]() { + srt_manager.join(); + std::lock_guard lock(queue->mutex); + queue->eof = true; + }); + + NamespaceMessage namespace_message{ + .draft = draft_version, + .track_namespace = track_namespace_, + .request_id = 0, + }; + if (draft_version == openmoq::publisher::DraftVersion::kDraft18) { + status = send_request_stream_and_wait( + transport_, draft_version, encode_namespace_message(namespace_message), false, nullptr, + &namespace_stream_id_); + } else { + status = write_frame(control_stream_id_, encode_namespace_message(namespace_message), false); + if (status.ok) { + status = collect_control_acknowledgements( + transport_, control_stream_id_, draft_version, 1, 0, pending_control_bytes_); + } + } + if (!status.ok) { + stop_requested = true; + if (srt_join_thread.joinable()) srt_join_thread.join(); + return status; + } + + std::map alias_by_track; + std::uint64_t next_alias = 0; + alias_by_track.emplace("catalog", next_alias++); + for (const auto& track : tracks) { + alias_by_track.emplace(track.track_name, next_alias++); + } + + bool catalog_sent = false; + auto send_catalog = [&](std::uint64_t track_alias) -> TransportStatus { + if (catalog_sent) { + return TransportStatus::success(); + } + const openmoq::publisher::CmsfObject catalog_object{ + .kind = openmoq::publisher::CmsfObjectKind::kInitialization, + .track_name = "catalog", + .group_id = 0, + .subgroup_id = 0, + .object_id = 0, + .media_time_us = 0, + .media_duration_us = 0, + .payload = {}, + .owned_payload = live_catalog.catalog_payload, + }; + SubgroupSenderState catalog_sender; + TransportStatus cat_status = catalog_sender.serve( + transport_, draft_version, track_alias, 0, + catalog_object, true, true, + std::span(live_catalog.catalog_payload)); + if (!cat_status.ok) { + return cat_status; + } + record_published_object("catalog", 0, live_catalog.catalog_payload.size()); + catalog_sent = true; + return TransportStatus::success(); + }; + + std::map sender_by_track; + std::map last_group_id_by_track; + std::map active_subscriptions; + std::set subscribed_tracks; + + auto drain_queue = [&]() -> TransportStatus { + while (true) { + openmoq::publisher::MediaFragment fragment; + { + std::lock_guard lock(queue->mutex); + if (queue->fragments.empty()) break; + fragment = std::move(queue->fragments.front()); + queue->fragments.pop_front(); + } + + if (!auto_forward_ && !subscribed_tracks.count(fragment.track_name)) { + continue; + } + + const auto alias_it = alias_by_track.find(fragment.track_name); + if (alias_it == alias_by_track.end()) { + continue; + } + + const openmoq::publisher::CmsfObject object{ + .kind = openmoq::publisher::CmsfObjectKind::kMedia, + .track_name = fragment.track_name, + .group_id = fragment.group_id, + .subgroup_id = 0, + .object_id = fragment.object_id, + .media_time_us = fragment.start_time_us, + .media_duration_us = fragment.duration_us, + .payload = {}, + .owned_payload = fragment.payload.owned_bytes, + }; + + auto& sender = sender_by_track[fragment.track_name]; + const auto group_it = last_group_id_by_track.find(fragment.track_name); + if (group_it != last_group_id_by_track.end() && group_it->second != static_cast(fragment.group_id)) { + TransportStatus finish_status = sender.finish_group(transport_); + if (!finish_status.ok) { + return finish_status; + } + } + + const std::uint64_t send_seq = next_send_seq(); + const std::span payload(fragment.payload.owned_bytes); + TransportStatus write_status = sender.serve( + transport_, draft_version, alias_it->second, send_seq, + object, true, false, payload); + if (!write_status.ok) { + return write_status; + } + last_group_id_by_track[fragment.track_name] = static_cast(fragment.group_id); + record_published_object(fragment.track_name, + static_cast(fragment.group_id), + fragment.payload.owned_bytes.size()); + if (publish_stats_.objects_published % 100 == 1) { + std::cerr << "[moqt-session] published track=" << fragment.track_name + << " group=" << fragment.group_id << " obj=" << fragment.object_id + << " bytes=" << fragment.payload.owned_bytes.size() + << " total_objects=" << publish_stats_.objects_published + << " total_groups=" << publish_stats_.groups_published << std::endl; + } + } + return TransportStatus::success(); + }; + + auto process_control_messages = [&]() -> std::pair { + std::size_t new_subs = 0; + std::size_t message_size = 0; + while (next_control_message(pending_control_bytes_, draft_version, message_size)) { + const std::vector message_bytes( + pending_control_bytes_.begin(), + pending_control_bytes_.begin() + static_cast(message_size)); + std::size_t offset = 0; + std::uint64_t message_type = 0; + if (!decode_varint(message_bytes, offset, message_type)) { + return {TransportStatus::failure("failed to parse control request type"), 0}; + } + + if (message_type == 0x03) { + SubscribeMessage subscribe; + if (!decode_subscribe_message(message_bytes, draft_version, subscribe)) { + return {TransportStatus::failure("received invalid SUBSCRIBE"), 0}; + } + + const auto track_it = alias_by_track.find(subscribe.track_name); + if (track_it == alias_by_track.end()) { + auto ws = transport_.write_stream(control_stream_id_, + encode_subscribe_error_message(subscribe.request_id, 0x2, "track does not exist"), false); + if (!ws.ok) { + return {ws, 0}; + } + } else { + auto ws = transport_.write_stream(control_stream_id_, + encode_subscribe_ok_message(draft_version, subscribe.request_id, + track_it->second, 0, 0, false), false); + if (!ws.ok) { + return {ws, 0}; + } + active_subscriptions.emplace(subscribe.request_id, subscribe); + ++new_subs; + + if (subscribe.track_name == "catalog") { + ws = send_catalog(track_it->second); + if (!ws.ok) { + return {ws, 0}; + } + ws = transport_.write_stream(control_stream_id_, + encode_publish_done_message(draft_version, subscribe.request_id, 1), false); + if (!ws.ok) { + return {ws, 0}; + } + } else { + subscribed_tracks.insert(subscribe.track_name); + } + } + } else if (message_type == 0x11) { + SubscribeNamespaceMessage subscribe_namespace; + if (decode_subscribe_namespace_message(message_bytes, draft_version, subscribe_namespace)) { + auto ws = transport_.write_stream(control_stream_id_, + encode_subscribe_namespace_ok_message(draft_version, subscribe_namespace.request_id), false); + if (!ws.ok) { + return {ws, 0}; + } + } + } + + pending_control_bytes_.erase( + pending_control_bytes_.begin(), + pending_control_bytes_.begin() + static_cast(message_size)); + } + return {TransportStatus::success(), new_subs}; + }; + + bool fin = false; + while (true) { + bool is_eof; + { + std::lock_guard lock(queue->mutex); + is_eof = queue->eof && queue->fragments.empty(); + } + + status = drain_queue(); + if (!status.ok) { + break; + } + + auto [ctrl_status, _new_subs] = process_control_messages(); + if (!ctrl_status.ok) { + status = ctrl_status; + break; + } + + std::vector chunk; + bool immediate_fin = false; + const TransportStatus read_status = + transport_.read_stream(control_stream_id_, chunk, immediate_fin, std::chrono::milliseconds(0)); + if (read_status.ok) { + pending_control_bytes_.insert(pending_control_bytes_.end(), chunk.begin(), chunk.end()); + fin = immediate_fin; + } + + if (fin || is_eof) { + break; + } + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + + stop_requested = true; + if (srt_join_thread.joinable()) { + srt_join_thread.join(); + } + + for (auto& [track_name, sender] : sender_by_track) { + TransportStatus finish_status = sender.finish_group(transport_); + if (!finish_status.ok && status.ok) { + status = finish_status; + } + static_cast(track_name); + } + + if (!status.ok) { + return status; + } + + for (const auto& [request_id, subscribe] : active_subscriptions) { + if (subscribe.track_name == "catalog") { + continue; + } + status = transport_.write_stream(control_stream_id_, + encode_publish_done_message(draft_version, request_id, sender_by_track[subscribe.track_name].stream_count()), false); + if (!status.ok) { + return status; + } + } + + return transport_.write_stream(control_stream_id_, + encode_publish_namespace_done_message(namespace_message), false); +} + TransportStatus MoqtSession::publish_live(std::istream& input, openmoq::publisher::DraftVersion draft_version, bool /*split_cmaf_chunks*/) { @@ -2979,12 +3331,16 @@ TransportStatus MoqtSession::publish_live(std::istream& input, alias_by_track.emplace(track.track_name, next_alias++); } + // Map PUBLISH request_ids to track names so we can handle PUBLISH_OK from + // relays that accept tracks via PUBLISH_OK without forwarding SUBSCRIBE. + std::map publish_request_id_to_track; + // Preannounce media tracks so relay implementations that need explicit // PUBLISH before subscribing can discover them. Do not wait for PUBLISH_OK // here: some relays first send SUBSCRIBE or stay idle until a subscriber // appears, and live await-subscribe mode must keep draining control bytes. if (!uses_request_streams(draft_version) && !auto_forward_) { - std::uint64_t pub_req_id = 2; // Publisher uses even request_ids per MOQT convention + std::uint64_t pub_req_id = 2; for (const auto& track : tracks) { const TrackMessage track_msg{ .draft = draft_version, @@ -3002,6 +3358,7 @@ TransportStatus MoqtSession::publish_live(std::istream& input, } std::cerr << "[moqt-session] live: PUBLISH track=" << track.track_name << " request_id=" << pub_req_id << '\n'; + publish_request_id_to_track.emplace(pub_req_id, track.track_name); pub_req_id += 2; } } @@ -3131,6 +3488,7 @@ TransportStatus MoqtSession::publish_live(std::istream& input, // Main loop: drain queue and publish fragments std::map sender_by_track; + std::map last_group_id_by_track; std::map active_subscriptions; std::map active_subscription_stream_ids; std::set subscribed_tracks; @@ -3175,18 +3533,24 @@ TransportStatus MoqtSession::publish_live(std::istream& input, .owned_payload = fragment.payload.owned_bytes, }; + auto& sender = sender_by_track[fragment.track_name]; + const auto group_it = last_group_id_by_track.find(fragment.track_name); + if (group_it != last_group_id_by_track.end() && group_it->second != static_cast(fragment.group_id)) { + TransportStatus finish_status = sender.finish_group(transport_); + if (!finish_status.ok) { + return finish_status; + } + } + const std::uint64_t send_seq = next_send_seq(); const std::span payload(fragment.payload.owned_bytes); - - // Each fragment is a single object; FIN the stream immediately. - // With keyframe-based grouping each group typically has only 1 object - // per track (since GOP interval = fragment duration). - TransportStatus write_status = sender_by_track[fragment.track_name].serve( + TransportStatus write_status = sender.serve( transport_, draft_version, alias_it->second, send_seq, object, true, true, payload); if (!write_status.ok) { return write_status; } + last_group_id_by_track[fragment.track_name] = static_cast(fragment.group_id); record_published_object(fragment.track_name, static_cast(fragment.group_id), fragment.payload.owned_bytes.size()); @@ -3508,6 +3872,21 @@ TransportStatus MoqtSession::publish_live(std::istream& input, return {ws, 0}; } } + } else if (message_type == 0x1e) { // PUBLISH_OK + // Relay accepted a PUBLISHed track. Some relays (e.g. moqx) do not + // forward SUBSCRIBE after PUBLISH_OK; they expect the publisher to + // start sending data once the track is accepted. Mark the track as + // subscribed so drain_queue will forward fragments. + PublishOk publish_ok_msg; + if (decode_publish_ok(message_bytes, draft_version, publish_ok_msg)) { + const auto it = publish_request_id_to_track.find(publish_ok_msg.request_id); + if (it != publish_request_id_to_track.end()) { + subscribed_tracks.insert(it->second); + ++new_subs; + std::cerr << "[moqt-session] live: PUBLISH_OK track=" << it->second + << " request_id=" << publish_ok_msg.request_id << '\n'; + } + } } // Skip other message types @@ -3676,6 +4055,14 @@ TransportStatus MoqtSession::publish_live(std::istream& input, stdin_thread.join(); + for (auto& [track_name, sender] : sender_by_track) { + status = sender.finish_group(transport_); + if (!status.ok) { + return status; + } + static_cast(track_name); + } + // Send PUBLISH_DONE for each subscribed media track (catalog already handled) for (const auto& [request_id, subscribe] : active_subscriptions) { if (subscribe.track_name == "catalog") { diff --git a/tests/cli_options_test.cpp b/tests/cli_options_test.cpp index a983678..7de4506 100644 --- a/tests/cli_options_test.cpp +++ b/tests/cli_options_test.cpp @@ -151,5 +151,26 @@ int main() { "expected stdin input source to avoid storing a file path"); } + // --live-source srt --srt-config /tmp/foo.json + { + const CliOptions options = parse({"openmoq-publisher", "--live-source", "srt", "--srt-config", "/tmp/foo.json", + "--endpoint", "localhost:4443", "--namespace", "ns"}); + ok &= expect(options.live_source == openmoq::publisher::LiveSourceKind::kSrt, + "expected --live-source srt"); + ok &= expect(options.srt_config_path == "/tmp/foo.json", + "expected --srt-config path"); + } + + // --live-source srt without --srt-config should fail + { + bool threw = false; + try { + parse({"openmoq-publisher", "--live-source", "srt", "--namespace", "ns"}); + } catch (...) { + threw = true; + } + ok &= expect(threw, "expected --live-source srt without --srt-config to fail"); + } + return ok ? 0 : 1; } diff --git a/tests/live_srt_config_test.cpp b/tests/live_srt_config_test.cpp new file mode 100644 index 0000000..fc5c0f7 --- /dev/null +++ b/tests/live_srt_config_test.cpp @@ -0,0 +1,70 @@ +#include "openmoq/publisher/live_srt_config.h" + +#include +#include +#include +#include + +namespace { + +bool expect(bool condition, const std::string& message) { + if (!condition) { + std::cerr << "FAIL: " << message << '\n'; + return false; + } + return true; +} + +} // namespace + +int main() { + bool ok = true; + + const std::filesystem::path config_path = + std::filesystem::temp_directory_path() / "openmoq-live-srt-config-test.json"; + + { + std::ofstream out(config_path); + out << R"({ + "srt_callers": [ + { + "id": "bbb", + "srt": { + "mode": "caller", + "host": "10.0.0.11", + "port": 9000, + "latency_ms": 120 + }, + "mpegts": { + "auto_detect_program": true, + "program_number": 1, + "video_pid": null, + "audio_pid": null + }, + "cmaf": { + "fragment_on_keyframe": true, + "empty_moov": true, + "default_base_moof": true, + "separate_moof_per_track": true, + "target_fragment_duration_ms": 1000 + } + } + ] +})"; + } + + const openmoq::publisher::LiveSrtConfig config = + openmoq::publisher::parse_live_srt_config_file(config_path); + ok &= expect(config.srt_callers.size() == 1, "expected one SRT caller"); + ok &= expect(config.srt_callers.front().id == "bbb", "expected SRT id"); + ok &= expect(config.srt_callers.front().srt.host == "10.0.0.11", "expected SRT host"); + ok &= expect(config.srt_callers.front().srt.port == 9000, "expected SRT port"); + ok &= expect(config.srt_callers.front().mpegts.program_number.has_value(), "expected program_number"); + ok &= expect(config.srt_callers.front().mpegts.program_number.value_or(0) == 1, + "expected program_number=1"); + + std::error_code ec; + std::filesystem::remove(config_path, ec); + + return ok ? 0 : 1; +}