Live fragmented MP4 publishing support

README.md

@@ -573,6 +573,30 @@ Practical notes:
 - if HEVC samples include in-band parameter sets, the publisher preserves `hev1` because rewriting those samples would be incorrect
 - if you start from a progressive MP4, this project can remux it internally, but pre-fragmented input is still the simpler and more efficient path
 
+### Live fragmented MP4 stdin publishing
+
+For live encoder pipelines, the publisher can consume fragmented MP4 directly from
+standard input.
+
+This live path expects ffmpeg to emit track-separated fragments, where each
+`moof` + `mdat` pair belongs to a single media track. Use `+separate_moof` when
+generating the stream. Without `+separate_moof`, audio and video may be carried
+inside the same `moof`, which is not the intended input layout for the current
+live parser.
+
+```bash
+ffmpeg -stream_loop -1 -re -i bbb_sunflower_1080p_30fps_normal.mp4 \
+  -map 0:v:0 -map 0:a:0 \
+  -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 \
+    --input - \
+    --endpoint moqt://relay.example.com:443/moq \
+    --namespace live/demo \
+    --timeout 120
+```
+
 ## CI
 
 GitHub Actions is configured to build and test the project on:

include/openmoq/publisher/cmaf_segmenter.h

@@ -25,6 +25,7 @@ struct MediaFragment {
     std::uint64_t duration_us = 0;
     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
     PayloadBuffer payload;
 };
 
@@ -38,4 +39,12 @@ SegmentedMp4 segment_for_cmaf(const ParsedMp4& parsed_mp4, CmafObjectMode object
 std::string summarize_tracks(const std::vector<TrackDescription>& tracks);
 std::size_t payload_size(const PayloadBuffer& payload);
 
+// Build a MediaFragment from a single moof+mdat pair for live streaming.
+// group_id is assigned by the caller (incremented per track).
+// The fragment owns the combined moof+mdat bytes.
+MediaFragment build_live_fragment(std::span<const std::uint8_t> moof_bytes,
+                                  std::span<const std::uint8_t> mdat_bytes,
+                                  const std::vector<TrackDescription>& tracks,
+                                  std::size_t group_id);
+
 }  // namespace openmoq::publisher

include/openmoq/publisher/cmsf_packager.h

@@ -48,4 +48,13 @@ void emit_plan_objects(const PublishPlan& plan,
                        std::span<const std::uint8_t> bytes,
                        const std::filesystem::path& output_dir);
 
+// Build catalog JSON and track-specific init segments for live streaming
+struct LiveCatalog {
+    std::vector<std::uint8_t> catalog_payload;
+    std::vector<TrackInitialization> track_initializations;
+};
+LiveCatalog build_live_catalog(const std::vector<TrackDescription>& tracks,
+                               std::span<const std::uint8_t> init_segment,
+                               bool is_live = true);
+
 }  // namespace openmoq::publisher

include/openmoq/publisher/mp4_box.h

@@ -3,6 +3,7 @@
 #include <cstddef>
 #include <cstdint>
 #include <iosfwd>
+#include <optional>
 #include <span>
 #include <string>
 #include <string_view>
@@ -57,4 +58,31 @@ std::vector<const Mp4Box*> find_boxes(const std::vector<Mp4Box>& boxes, std::str
 const Mp4Box* find_child_box(const Mp4Box& box, std::string_view type);
 std::span<const std::uint8_t> slice_bytes(std::span<const std::uint8_t> bytes, const ByteSpan& span);
 
+// Incremental MP4 box reader for streaming input (e.g. piped ffmpeg).
+// Buffers raw bytes and yields complete top-level boxes one at a time.
+struct StreamingBoxResult {
+    std::string type;
+    std::vector<std::uint8_t> bytes;
+};
+
+class StreamingMp4Reader {
+public:
+    // Append raw data to internal buffer.
+    void append(const std::uint8_t* data, std::size_t len);
+
+    // Read up to chunk_size bytes from input and append.
+    // Returns number of bytes read; 0 means EOF.
+    std::size_t read_from(std::istream& input, std::size_t chunk_size = 16384);
+
+    // Try to extract the next complete top-level box from the buffer.
+    // Returns std::nullopt if not enough data is available yet.
+    std::optional<StreamingBoxResult> next_box();
+
+private:
+    std::vector<std::uint8_t> buffer_;
+    std::size_t consumed_ = 0;
+
+    void compact();
+};
+
 }  // namespace openmoq::publisher

include/openmoq/publisher/transport/moqt_session.h

@@ -3,6 +3,7 @@
 #include "openmoq/publisher/cmsf_packager.h"
 #include "openmoq/publisher/transport/publisher_transport.h"
 
+#include <iosfwd>
 #include <optional>
 #include <chrono>
 #include <span>
@@ -31,6 +32,9 @@ class MoqtSession {
 
     TransportStatus connect(const EndpointConfig& endpoint, const TlsConfig& tls);
     TransportStatus publish(const openmoq::publisher::PublishPlan& plan);
+    TransportStatus publish_live(std::istream& input,
+                                 openmoq::publisher::DraftVersion draft_version,
+                                 bool split_cmaf_chunks);
     TransportStatus close(std::uint64_t application_error_code = 0);
 
 private:

src/cmaf_segmenter.cpp

@@ -1005,4 +1005,149 @@ std::size_t payload_size(const PayloadBuffer& payload) {
     return payload.owned_bytes.empty() ? payload.span.size : payload.owned_bytes.size();
 }
 
+MediaFragment build_live_fragment(std::span<const std::uint8_t> moof_bytes,
+                                  std::span<const std::uint8_t> mdat_bytes,
+                                  const std::vector<TrackDescription>& tracks,
+                                  std::size_t group_id) {
+    // Extract timing data directly from the standalone moof bytes.
+    // For fragmented MP4 from ffmpeg, tfhd carries default_sample_flags so we
+    // do not need to look up trex defaults from the init segment.
+    const std::vector<Mp4Box> moof_boxes = parse_mp4_boxes(moof_bytes);
+    if (moof_boxes.empty() || moof_boxes.front().type != "moof") {
+        throw std::runtime_error("build_live_fragment: expected moof box");
+    }
+    const Mp4Box& moof = moof_boxes.front();
+
+    // Extract track name from moof -> traf -> tfhd -> track_id.
+    const std::string track_name = fragment_track_name(moof, tracks, moof_bytes);
+
+    // Extract timing information directly from moof bytes.
+    const TrackDescription* track_desc = fragment_track_description(moof, tracks, moof_bytes);
+    if (track_desc == nullptr || track_desc->timescale == 0) {
+        throw std::runtime_error("build_live_fragment: cannot find track for fragment");
+    }
+
+    const Mp4Box* traf = find_child_box(moof, "traf");
+    if (traf == nullptr) {
+        throw std::runtime_error("build_live_fragment: moof has no traf");
+    }
+
+    std::uint64_t base_decode_time = 0;
+    if (const Mp4Box* tfdt = find_child_box(*traf, "tfdt")) {
+        const std::uint8_t version = moof_bytes[tfdt->payload.offset];
+        const std::size_t time_offset = tfdt->payload.offset + 4;
+        if (version == 1 && time_offset + 8 <= moof_bytes.size()) {
+            std::uint64_t val = 0;
+            for (int i = 0; i < 8; ++i) {
+                val = (val << 8U) | moof_bytes[time_offset + i];
+            }
+            base_decode_time = val;
+        } else if (time_offset + 4 <= moof_bytes.size()) {
+            base_decode_time = read_be32(moof_bytes, time_offset);
+        }
+    }
+
+    std::uint32_t default_sample_duration = 0;
+    std::uint32_t default_sample_flags = 0x02000000U;
+    if (const Mp4Box* tfhd = find_child_box(*traf, "tfhd")) {
+        const std::uint32_t flags = read_full_box_flags(*tfhd, moof_bytes);
+        std::size_t cursor = tfhd->payload.offset + 8;
+        if ((flags & 0x000001U) != 0) cursor += 8;
+        if ((flags & 0x000002U) != 0) cursor += 4;
+        if ((flags & 0x000008U) != 0 && cursor + 4 <= moof_bytes.size()) {
+            default_sample_duration = read_be32(moof_bytes, cursor);
+            cursor += 4;
+        }
+        if ((flags & 0x000010U) != 0 && cursor + 4 <= moof_bytes.size()) {
+            cursor += 4;
+        }
+        if ((flags & 0x000020U) != 0 && cursor + 4 <= moof_bytes.size()) {
+            default_sample_flags = read_be32(moof_bytes, cursor);
+        }
+    }
+
+    std::uint64_t duration = 0;
+    std::uint64_t earliest_presentation_time = base_decode_time;
+    bool earliest_presentation_time_set = false;
+    std::uint32_t first_sample_flags = default_sample_flags;
+    if (const Mp4Box* trun = find_child_box(*traf, "trun")) {
+        const std::uint32_t flags = read_full_box_flags(*trun, moof_bytes);
+        std::size_t cursor = trun->payload.offset + 4;
+        if (cursor + 4 <= moof_bytes.size()) {
+            const std::uint32_t sample_count = read_be32(moof_bytes, cursor);
+            cursor += 4;
+            if ((flags & 0x000001U) != 0) cursor += 4;
+            bool first_sample_flags_present = false;
+            if ((flags & 0x000004U) != 0) {
+                first_sample_flags = read_be32_or_zero(moof_bytes, cursor);
+                first_sample_flags_present = true;
+                cursor += 4;
+            }
+            std::uint64_t sample_decode_time = base_decode_time;
+            for (std::uint32_t i = 0; i < sample_count && cursor <= moof_bytes.size(); ++i) {
+                std::uint32_t sample_duration = default_sample_duration;
+                if ((flags & 0x000100U) != 0 && cursor + 4 <= moof_bytes.size()) {
+                    sample_duration = read_be32(moof_bytes, cursor);
+                    cursor += 4;
+                }
+
+                if ((flags & 0x000200U) != 0) cursor += 4;  // skip sample_size
+                std::uint32_t sample_flags = first_sample_flags_present && i == 0
+                                                 ? first_sample_flags : default_sample_flags;
+                if ((flags & 0x000400U) != 0 && cursor + 4 <= moof_bytes.size()) {
+                    sample_flags = read_be32(moof_bytes, cursor);
+                    cursor += 4;
+                }
+                std::int32_t composition_offset = 0;
+                if ((flags & 0x000800U) != 0 && cursor + 4 <= moof_bytes.size()) {
+                    composition_offset = static_cast<std::int32_t>(read_be32(moof_bytes, cursor));
+                    cursor += 4;
+                }
+                const std::int64_t pt_signed =
+                    static_cast<std::int64_t>(sample_decode_time) + static_cast<std::int64_t>(composition_offset);
+                const std::uint64_t pt = pt_signed < 0 ? 0 : static_cast<std::uint64_t>(pt_signed);
+                if (!earliest_presentation_time_set || pt < earliest_presentation_time) {
+                    earliest_presentation_time = pt;
+                    earliest_presentation_time_set = true;
+                }
+                if (i == 0 && (flags & 0x000400U) != 0) {
+                    first_sample_flags = sample_flags;
+                }
+                duration += sample_duration;
+                sample_decode_time += sample_duration;
+            }
+        }
+    }
+
+    std::uint8_t sap_type = 0;
+    const bool first_sample_is_sync = (first_sample_flags & 0x00010000U) == 0;
+    const bool is_video = (track_desc->handler_type == "vide");
+    if (!is_video) {
+        sap_type = 1;
+    } else if (first_sample_is_sync) {
+        sap_type = 2;
+    }
+
+    // A video keyframe: video track with sync first sample
+    const bool is_video_keyframe = is_video && first_sample_is_sync;
+
+    // Combine moof+mdat into a single owned payload (CMSF compliance).
+    std::vector<std::uint8_t> payload;
+    payload.reserve(moof_bytes.size() + mdat_bytes.size());
+    payload.insert(payload.end(), moof_bytes.begin(), moof_bytes.end());
+    payload.insert(payload.end(), mdat_bytes.begin(), mdat_bytes.end());
+
+    return MediaFragment{
+        .group_id = group_id,
+        .object_id = 0,
+        .track_name = track_name,
+        .start_time_us = scale_to_us(base_decode_time, track_desc->timescale),
+        .duration_us = scale_to_us(duration, track_desc->timescale),
+        .earliest_presentation_time_us = scale_to_us(earliest_presentation_time, track_desc->timescale),
+        .sap_type = sap_type,
+        .is_video_keyframe = is_video_keyframe,
+        .payload = {.span = {}, .owned_bytes = std::move(payload)},
+    };
+}
+
 }  // namespace openmoq::publisher

src/cmsf_packager.cpp

@@ -609,4 +609,104 @@ void emit_plan_objects(const PublishPlan& plan,
     manifest << render_publish_plan(plan);
 }
 
+LiveCatalog build_live_catalog(const std::vector<TrackDescription>& tracks,
+                               std::span<const std::uint8_t> init_segment,
+                               bool is_live) {
+    // Local base64 encoder
+    auto local_base64_encode = [](std::span<const std::uint8_t> bytes) -> std::string {
+        static const char kAlphabet[] = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
+        std::string result;
+        result.reserve(((bytes.size() + 2) / 3) * 4);
+        for (std::size_t i = 0; i < bytes.size(); i += 3) {
+            const std::uint32_t b0 = bytes[i];
+            const std::uint32_t b1 = i + 1 < bytes.size() ? bytes[i + 1] : 0;
+            const std::uint32_t b2 = i + 2 < bytes.size() ? bytes[i + 2] : 0;
+            result.push_back(kAlphabet[b0 >> 2]);
+            result.push_back(kAlphabet[((b0 & 0x3) << 4) | (b1 >> 4)]);
+            result.push_back(i + 1 < bytes.size() ? kAlphabet[((b1 & 0xf) << 2) | (b2 >> 6)] : '=');
+            result.push_back(i + 2 < bytes.size() ? kAlphabet[b2 & 0x3f] : '=');
+        }
+        return result;
+    };
+
+    auto local_json_escape = [](std::string_view s) -> std::string {
+        std::string result;
+        result.reserve(s.size());
+        for (char c : s) {
+            switch (c) {
+                case '"': result += "\\\""; break;
+                case '\\': result += "\\\\"; break;
+                case '\n': result += "\\n"; break;
+                case '\r': result += "\\r"; break;
+                case '\t': result += "\\t"; break;
+                default: result += c; break;
+            }
+        }
+        return result;
+    };
+
+    auto local_track_role = [](std::string_view handler_type) -> std::string_view {
+        if (handler_type == "vide") return "video";
+        if (handler_type == "soun") return "audio";
+        if (handler_type == "meta") return "data";
+        return "data";
+    };
+
+    LiveCatalog result;
+
+    // For live streaming, use full init segment for all tracks (simpler than building track-specific)
+    const std::string full_init_base64 = local_base64_encode(init_segment);
+    for (const auto& track : tracks) {
+        result.track_initializations.push_back({
+            .track_name = track.track_name,
+            .codec_payload = {},
+            .init_segment = std::vector<std::uint8_t>(init_segment.begin(), init_segment.end()),
+        });
+    }
+
+    // Build catalog JSON
+    std::ostringstream catalog;
+    catalog << "{";
+    catalog << "\"version\":1,";
+    catalog << "\"format\":\"cmsf\",";
+    catalog << "\"tracks\":[";
+    bool first_track = true;
+    for (const auto& track : tracks) {
+        if (!first_track) {
+            catalog << ',';
+        }
+        first_track = false;
+
+        catalog << '{'
+                << "\"name\":\"" << local_json_escape(track.track_name) << "\","
+                << "\"id\":" << track.track_id << ','
+                << "\"role\":\"" << local_track_role(track.handler_type) << "\","
+                << "\"packaging\":\"" << local_json_escape(track.packaging) << "\","
+                << "\"renderGroup\":1,"
+                << "\"isLive\":" << (is_live ? "true" : "false");
+        if (!track.codec.empty()) {
+            catalog << ",\"codec\":\"" << local_json_escape(track.codec) << '"';
+        }
+        if (track.handler_type == "vide") {
+            catalog << ",\"width\":" << track.width
+                    << ",\"height\":" << track.height;
+            if (track.frame_rate > 0.0) {
+                catalog << std::fixed << std::setprecision(2) << ",\"frameRate\":" << track.frame_rate;
+            }
+        } else if (track.handler_type == "soun") {
+            catalog << ",\"sampleRate\":" << track.sample_rate
+                    << ",\"channelCount\":" << track.channel_count;
+        }
+        // Use full init segment for all tracks
+        catalog << ",\"initData\":\"" << full_init_base64 << '"';
+        catalog << '}';
+    }
+    catalog << "]}";
+
+    const std::string catalog_text = catalog.str();
+    result.catalog_payload = std::vector<std::uint8_t>(catalog_text.begin(), catalog_text.end());
+
+    return result;
+}
+
 }  // namespace openmoq::publisher