Support file and stdin MP4 input

Author: mondain

README.md

@@ -157,12 +157,35 @@ OPENMOQ_PICOQUIC_TRACE=1 ./build/openmoq-publisher-picoquic-smoke-tests
 
 ### CLI dry-run testing
 
+Input source note:
+
+- `--input <path>` reads an MP4 from a regular file
+- `--input -` reads the MP4 byte stream from standard input, which allows `cat`, `ffmpeg`, or other producer pipelines to feed the publisher directly
+- fragmented and progressive MP4 inputs are both supported through either source type
+
 Use `--dump-plan` to inspect the generated publish plan without touching the network:
 
 ```bash
 ./build/openmoq-publisher --input sample.mp4 --draft 14 --dump-plan
 ```
 
+Use stdin when the source is already being produced by another command:
+
+```bash
+cat sample.mp4 | ./build/openmoq-publisher --input - --draft 14 --dump-plan
+```
+
+For example, you can inspect an ffmpeg-produced fragmented stream without writing an intermediate file:
+
+```bash
+ffmpeg -i input.mp4 \
+  -map 0:v -map 0:a \
+  -c:v copy \
+  -c:a copy \
+  -movflags +frag_keyframe+empty_moov+default_base_moof+separate_moof \
+  -f mp4 - | ./build/openmoq-publisher --input - --draft 14 --dump-plan
+```
+
 Use `--emit-dir` to inspect the emitted catalog and media objects on disk:
 
 ```bash
@@ -274,6 +297,19 @@ Transport-oriented CLI flags are also present now:
   --insecure
 ```
 
+The same CLI accepts stdin for transport publishing as well:
+
+```bash
+cat sample.mp4 | ./build/openmoq-publisher \
+  --input - \
+  --endpoint localhost:4433 \
+  --namespace media \
+  --forward 0 \
+  --timeout 3 \
+  --paced \
+  --insecure
+```
+
 Chunk/object mapping:
 
 - default behavior is lower-latency split publication, which emits multiple MOQT objects in the same group when CMAF chunk/sample boundaries are available

include/openmoq/publisher/cli_options.h

@@ -10,8 +10,18 @@
 
 namespace openmoq::publisher {
 
+enum class InputSourceKind {
+    kFile,
+    kStdin,
+};
+
+struct InputSource {
+    InputSourceKind kind = InputSourceKind::kFile;
+    std::filesystem::path path;
+};
+
 struct CliOptions {
-    std::filesystem::path input_path;
+    InputSource input_source;
     std::optional<std::filesystem::path> emit_dir;
     std::optional<transport::EndpointConfig> endpoint;
     transport::TlsConfig tls;

include/openmoq/publisher/mp4_box.h

@@ -2,6 +2,7 @@
 
 #include <cstddef>
 #include <cstdint>
+#include <iosfwd>
 #include <span>
 #include <string>
 #include <string_view>
@@ -46,6 +47,7 @@ struct ParsedMp4 {
 };
 
 ParsedMp4 parse_mp4_file(const std::string& path);
+ParsedMp4 parse_mp4_stream(std::istream& input, std::string_view source_name);
 std::vector<Mp4Box> parse_mp4_boxes(std::span<const std::uint8_t> bytes);
 std::vector<TrackDescription> extract_tracks(const std::vector<Mp4Box>& top_level_boxes,
                                              std::span<const std::uint8_t> bytes);

src/cli_options.cpp

@@ -7,6 +7,13 @@ namespace openmoq::publisher {
 
 namespace {
 
+InputSource parse_input_source(std::string_view value) {
+    if (value == "-") {
+        return InputSource{.kind = InputSourceKind::kStdin, .path = {}};
+    }
+    return InputSource{.kind = InputSourceKind::kFile, .path = std::filesystem::path(value)};
+}
+
 DraftVersion parse_draft(std::string_view value) {
     if (value == "14") {
         return DraftVersion::kDraft14;
@@ -84,7 +91,7 @@ CliOptions parse_cli_options(int argc, char** argv) {
         };
 
         if (argument == "--input") {
-            options.input_path = require_value("--input");
+            options.input_source = parse_input_source(require_value("--input"));
         } else if (argument == "--endpoint") {
             options.endpoint = parse_endpoint(require_value("--endpoint"));
         } else if (argument == "--alpn") {
@@ -131,7 +138,7 @@ CliOptions parse_cli_options(int argc, char** argv) {
         }
     }
 
-    if (options.input_path.empty()) {
+    if (options.input_source.kind == InputSourceKind::kFile && options.input_source.path.empty()) {
         throw std::runtime_error("missing required --input argument");
     }
 
@@ -147,7 +154,7 @@ CliOptions parse_cli_options(int argc, char** argv) {
 
 std::string build_usage(const char* argv0) {
     return std::string("Usage: ") + argv0 +
-           " --input <mp4> [--draft 14|16] [--namespace <value>] [--forward 0|1] [--timeout <seconds>]"
+           " --input <mp4|-> [--draft 14|16] [--namespace <value>] [--forward 0|1] [--timeout <seconds>]"
            " [--publish-catalog] [--coalesce-cmaf-chunks|--coalesce-cmaf-chunk] [--paced] [--dump-plan] [--emit-dir <dir>]"
            " [--endpoint host:port|moqt://host:port/path] [--alpn value] [--sni value]"
            " [--cert file] [--key file] [--ca file] [--insecure]";

src/main.cpp

@@ -13,7 +13,9 @@ int main(int argc, char** argv) {
 
     try {
         const CliOptions options = parse_cli_options(argc, argv);
-        const ParsedMp4 parsed_mp4 = parse_mp4_file(options.input_path.string());
+        const ParsedMp4 parsed_mp4 = options.input_source.kind == InputSourceKind::kStdin
+                                         ? parse_mp4_stream(std::cin, "stdin")
+                                         : parse_mp4_file(options.input_source.path.string());
         const SegmentedMp4 segmented_mp4 = segment_for_cmaf(parsed_mp4,
                                                             options.split_cmaf_chunks ? CmafObjectMode::kSplit
                                                                                       : CmafObjectMode::kCoalesced);

src/mp4_box.cpp

@@ -285,16 +285,43 @@ ParsedMp4 parse_mp4_file(const std::string& path) {
         throw std::runtime_error("failed to open MP4 file: " + path);
     }
 
-    input.seekg(0, std::ios::end);
-    const auto size = static_cast<std::size_t>(input.tellg());
-    input.seekg(0, std::ios::beg);
+    return parse_mp4_stream(input, path);
+}
 
+ParsedMp4 parse_mp4_stream(std::istream& input, std::string_view source_name) {
     ParsedMp4 parsed;
-    parsed.bytes.resize(size);
-    input.read(reinterpret_cast<char*>(parsed.bytes.data()), static_cast<std::streamsize>(size));
 
-    if (!input) {
-        throw std::runtime_error("failed to read MP4 file: " + path);
+    input.seekg(0, std::ios::end);
+    if (input.good()) {
+        const auto end = input.tellg();
+        if (end >= 0) {
+            parsed.bytes.resize(static_cast<std::size_t>(end));
+            input.seekg(0, std::ios::beg);
+            input.read(reinterpret_cast<char*>(parsed.bytes.data()), static_cast<std::streamsize>(parsed.bytes.size()));
+            if (!input) {
+                throw std::runtime_error("failed to read MP4 input: " + std::string(source_name));
+            }
+        }
+    }
+
+    if (parsed.bytes.empty()) {
+        input.clear();
+        input.seekg(0, std::ios::beg);
+
+        constexpr std::size_t kChunkSize = 16 * 1024;
+        std::array<char, kChunkSize> buffer{};
+        while (input) {
+            input.read(buffer.data(), static_cast<std::streamsize>(buffer.size()));
+            const auto bytes_read = input.gcount();
+            if (bytes_read > 0) {
+                parsed.bytes.insert(parsed.bytes.end(),
+                                    reinterpret_cast<const std::uint8_t*>(buffer.data()),
+                                    reinterpret_cast<const std::uint8_t*>(buffer.data()) + bytes_read);
+            }
+        }
+        if (!input.eof()) {
+            throw std::runtime_error("failed to read MP4 input: " + std::string(source_name));
+        }
     }
 
     parsed.top_level_boxes = parse_mp4_boxes(parsed.bytes);

tests/cli_options_test.cpp

@@ -38,6 +38,10 @@ int main() {
         ok &= expect(options.subscriber_timeout == std::chrono::seconds(3),
                      "expected default subscriber timeout to remain 3 seconds");
         ok &= expect(options.split_cmaf_chunks, "expected chunk splitting to be enabled by default");
+        ok &= expect(options.input_source.kind == openmoq::publisher::InputSourceKind::kFile,
+                     "expected file input to remain the default input source kind");
+        ok &= expect(options.input_source.path == "sample.mp4",
+                     "expected file input path to be preserved");
     }
 
     {
@@ -93,5 +97,13 @@ int main() {
         ok &= expect(threw, "expected negative --timeout to be rejected");
     }
 
+    {
+        const CliOptions options = parse({"openmoq-publisher", "--input", "-"});
+        ok &= expect(options.input_source.kind == openmoq::publisher::InputSourceKind::kStdin,
+                     "expected --input - to select stdin");
+        ok &= expect(options.input_source.path.empty(),
+                     "expected stdin input source to avoid storing a file path");
+    }
+
     return ok ? 0 : 1;
 }

tests/cmaf_segmenter_test.cpp

@@ -5,6 +5,7 @@
 #include <cstdint>
 #include <iostream>
 #include <map>
+#include <sstream>
 #include <string>
 #include <vector>
 
@@ -333,6 +334,15 @@ int main() {
     ok &= expect(payload_size(segmented.initialization_segment) > 0, "expected fragmented init payload");
     ok &= expect(payload_size(segmented.fragments.front().payload) > 0, "expected fragmented media payload");
 
+    std::stringstream fragmented_stream(std::ios::in | std::ios::out | std::ios::binary);
+    fragmented_stream.write(reinterpret_cast<const char*>(fragmented_bytes.data()),
+                            static_cast<std::streamsize>(fragmented_bytes.size()));
+    fragmented_stream.seekg(0, std::ios::beg);
+    const ParsedMp4 parsed_from_stream = parse_mp4_stream(fragmented_stream, "memory");
+    ok &= expect(parsed_from_stream.bytes == fragmented_bytes, "expected stream parser to preserve input bytes");
+    ok &= expect(parsed_from_stream.top_level_boxes.size() == 4, "expected stream parser to decode top-level boxes");
+    ok &= expect(parsed_from_stream.tracks.size() == 1, "expected stream parser to extract track metadata");
+
     const auto multitrack_fragmented_bytes = make_multitrack_fragmented_test_mp4();
     ParsedMp4 multitrack_fragmented{
         .bytes = multitrack_fragmented_bytes,