Merge pull request #18 from mondain/feat/switch-stream-per-obj

added --stream-per-object flag

Author: mondain

include/openmoq/publisher/cli_options.h

@@ -42,6 +42,7 @@ struct CliOptions {
     bool include_sap = false;
     bool include_msf_timeline = false;
     bool split_cmaf_chunks = true;
+    bool stream_per_object = false;
     bool paced = false;
     bool loop = false;
     bool dump_plan = false;

include/openmoq/publisher/publisher_api.h

@@ -28,6 +28,7 @@ struct PublisherConfig {
     bool include_sap = false;
     bool include_msf_timeline = false;
     bool split_cmaf_chunks = true;
+    bool live_stream_per_object = false;
     bool paced = false;
     bool loop = false;
     std::chrono::seconds subscriber_timeout = std::chrono::seconds(30);

include/openmoq/publisher/transport/moqt_session.h

@@ -67,11 +67,13 @@ class MoqtSession {
     TransportStatus publish(const openmoq::publisher::PublishPlan& plan);
     TransportStatus publish_live(std::istream& input,
                                  openmoq::publisher::DraftVersion draft_version,
-                                 bool split_cmaf_chunks);
+                                 bool split_cmaf_chunks,
+                                 bool stream_per_object = false);
     TransportStatus publish_live(const LiveIngestOptions& ingest,
                                  std::istream* stdin_input,
                                  openmoq::publisher::DraftVersion draft_version,
-                                 bool split_cmaf_chunks);
+                                 bool split_cmaf_chunks,
+                                 bool stream_per_object = false);
     TransportStatus publish_live_objects(const openmoq::publisher::LiveObjectSource& source,
                                          openmoq::publisher::DraftVersion draft_version);
     TransportStatus close(std::uint64_t application_error_code = 0);

src/cli_options.cpp

@@ -178,6 +178,8 @@ CliOptions parse_cli_options(int argc, char** argv) {
             options.include_msf_timeline = true;
         } else if (argument == "--coalesce-cmaf-chunks") {
             options.split_cmaf_chunks = false;
+        } else if (argument == "--stream-per-object") {
+            options.stream_per_object = true;
         } else if (argument == "--timeout") {
             options.subscriber_timeout = parse_timeout(require_value("--timeout"));
         } else if (argument == "--paced") {
@@ -241,7 +243,7 @@ std::string build_usage(const char* argv0) {
     return std::string("Usage: ") + argv0 +
            " --input <mp4|-> [--live-source auto|stdin|srt] [--srt-config <path>]"
            " [--transport raw|webtransport] [--draft 14|16|17|18] [--namespace <value>] [--forward 0|1] [--timeout <seconds>]"
-           " [--publish-catalog] [--sap] [--msf-timeline] [--coalesce-cmaf-chunks] [--paced] [--loop] [--dump-plan] [--emit-dir <dir>]"
+            " [--publish-catalog] [--sap] [--msf-timeline] [--coalesce-cmaf-chunks] [--stream-per-object] [--paced] [--loop] [--dump-plan] [--emit-dir <dir>]"
            " [--endpoint host:port|moqt://host:port/path|https://host:port/path] [--alpn value] [--sni value]"
            " [--cert file] [--key file] [--ca file] [--insecure]";
 }

src/main.cpp

@@ -30,6 +30,7 @@ int main(int argc, char** argv) {
             .include_sap = options.include_sap,
             .include_msf_timeline = options.include_msf_timeline,
             .split_cmaf_chunks = options.split_cmaf_chunks,
+            .live_stream_per_object = options.stream_per_object,
             .paced = options.paced,
             .loop = options.loop,
             .subscriber_timeout = options.subscriber_timeout,

src/publisher_api.cpp

@@ -256,7 +256,8 @@ transport::TransportStatus Publisher::publish_live(const LiveIngestConfig& inges
     status = active->session->publish_live(session_ingest,
                                            stdin_input,
                                            config_.draft_version,
-                                           config_.split_cmaf_chunks);
+                                           config_.split_cmaf_chunks,
+                                           config_.live_stream_per_object);
     if (!status.ok) {
         const std::string error = "transport live publish failed: " + status.message;
         static_cast<void>(active->session->close(0));

src/transport/moqt_session.cpp

@@ -2930,7 +2930,8 @@ TransportStatus MoqtSession::publish(const openmoq::publisher::PublishPlan& plan
 TransportStatus MoqtSession::publish_live(const LiveIngestOptions& ingest,
                                           std::istream* stdin_input,
                                           openmoq::publisher::DraftVersion draft_version,
-                                          bool split_cmaf_chunks) {
+                                          bool split_cmaf_chunks,
+                                          bool stream_per_object) {
     if (ingest.use_stdin && !ingest.srt_callers.empty()) {
         return TransportStatus::failure("mixed stdin+SRT ingest is not supported; use either stdin or srt");
     }
@@ -2941,7 +2942,7 @@ TransportStatus MoqtSession::publish_live(const LiveIngestOptions& ingest,
         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);
+        return publish_live(*stdin_input, draft_version, split_cmaf_chunks, stream_per_object);
     }
 
     // SRT-only path below.
@@ -3281,7 +3282,8 @@ TransportStatus MoqtSession::publish_live(const LiveIngestOptions& ingest,
 
 TransportStatus MoqtSession::publish_live(std::istream& input,
                                            openmoq::publisher::DraftVersion draft_version,
-                                           bool /*split_cmaf_chunks*/) {
+                                           bool /*split_cmaf_chunks*/,
+                                           bool stream_per_object) {
     if (transport_.state() != ConnectionState::kConnected) {
         return TransportStatus::failure("transport is not connected");
     }
@@ -3590,7 +3592,7 @@ TransportStatus MoqtSession::publish_live(std::istream& input,
             const std::span<const std::uint8_t> payload(fragment.payload.owned_bytes);
             TransportStatus write_status = sender.serve(
                 transport_, draft_version, alias_it->second, send_seq,
-                object, true, true, payload);
+                object, true, stream_per_object, payload);
             if (!write_status.ok) {
                 return write_status;
             }