From 2f0e0bdfcb08ed33a05c936788fb951c2f50e78a Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Thu, 18 Jun 2026 21:13:48 +0100 Subject: [PATCH] Share Muninn media feedback wire encoder --- native/obs_stem_source/CMakeLists.txt | 6 + .../mimir_obs_muninn_source.cpp | 228 +---------------- native/obs_stem_source/muninn_media_wire.h | 235 ++++++++++++++++++ tools/muninn_media_decoder_probe.cpp | 55 +++- 4 files changed, 295 insertions(+), 229 deletions(-) create mode 100644 native/obs_stem_source/muninn_media_wire.h diff --git a/native/obs_stem_source/CMakeLists.txt b/native/obs_stem_source/CMakeLists.txt index 4a64232..c56063e 100644 --- a/native/obs_stem_source/CMakeLists.txt +++ b/native/obs_stem_source/CMakeLists.txt @@ -15,3 +15,9 @@ target_compile_definitions(mimir_obs_stem_source PRIVATE NOMINMAX _WIN32_WINNT=0x0A00) target_link_libraries(mimir_obs_stem_source PRIVATE OBS::libobs d3d11 d3d12 dxgi ws2_32) + +add_executable(muninn_media_decoder_probe + ../../tools/muninn_media_decoder_probe.cpp) +target_compile_features(muninn_media_decoder_probe PRIVATE cxx_std_20) +target_include_directories(muninn_media_decoder_probe PRIVATE + ${CMAKE_CURRENT_SOURCE_DIR}) diff --git a/native/obs_stem_source/mimir_obs_muninn_source.cpp b/native/obs_stem_source/mimir_obs_muninn_source.cpp index 0b0d17e..a1b2aa6 100644 --- a/native/obs_stem_source/mimir_obs_muninn_source.cpp +++ b/native/obs_stem_source/mimir_obs_muninn_source.cpp @@ -6,10 +6,11 @@ #include +#include "muninn_media_wire.h" + #include #include #include -#include #include #include #include @@ -38,7 +39,6 @@ constexpr uint16_t MuninnDefaultMediaPort = 5204; constexpr const char *MuninnObsCatalogType = "muninn.obs_stream_catalog"; constexpr const char *MuninnObsCatalogSchema = "muninn.obs_stream_catalog.v1"; constexpr const char *MuninnObsCatalogKey = "obs"; -constexpr const char *MuninnMediaReceiverFeedbackSchema = "muninn.media_receiver_feedback.v1"; constexpr uint32_t MuninnMediaRudpConnectionId = 0x6d750001; constexpr auto MuninnVideoAssemblyDeadline = std::chrono::milliseconds(75); @@ -432,228 +432,6 @@ private: } }; -class MediaMsgpackWriter { -public: - const std::vector &bytes() const { return bytes_; } - - void write_raw(const std::vector &value) - { - bytes_.insert(bytes_.end(), value.begin(), value.end()); - } - - void write_nil() { bytes_.push_back(0xc0); } - - void write_bool(bool value) { bytes_.push_back(value ? 0xc3 : 0xc2); } - - void write_u64(uint64_t value) - { - if (value <= 0x7f) { - bytes_.push_back(static_cast(value)); - } else if (value <= 0xff) { - bytes_.push_back(0xcc); - bytes_.push_back(static_cast(value)); - } else if (value <= 0xffff) { - bytes_.push_back(0xcd); - write_be16_value(static_cast(value)); - } else if (value <= 0xffffffffull) { - bytes_.push_back(0xce); - write_be32_value(static_cast(value)); - } else { - bytes_.push_back(0xcf); - write_be64_value(value); - } - } - - void write_i64(int64_t value) - { - if (value >= 0) { - write_u64(static_cast(value)); - return; - } - if (value >= -32) { - bytes_.push_back(static_cast(0xe0 | (value + 32))); - } else if (value >= INT8_MIN) { - bytes_.push_back(0xd0); - bytes_.push_back(static_cast(value)); - } else if (value >= INT16_MIN) { - bytes_.push_back(0xd1); - write_be16_value(static_cast(value)); - } else if (value >= INT32_MIN) { - bytes_.push_back(0xd2); - write_be32_value(static_cast(value)); - } else { - bytes_.push_back(0xd3); - write_be64_value(static_cast(value)); - } - } - - void write_string(const std::string &value) - { - const auto length = value.size(); - if (length <= 31) { - bytes_.push_back(static_cast(0xa0 | length)); - } else if (length <= 0xff) { - bytes_.push_back(0xd9); - bytes_.push_back(static_cast(length)); - } else if (length <= 0xffff) { - bytes_.push_back(0xda); - write_be16_value(static_cast(length)); - } else { - bytes_.push_back(0xdb); - write_be32_value(static_cast(length)); - } - bytes_.insert(bytes_.end(), value.begin(), value.end()); - } - - void write_binary(const std::vector &value) - { - const auto length = value.size(); - if (length <= 0xff) { - bytes_.push_back(0xc4); - bytes_.push_back(static_cast(length)); - } else if (length <= 0xffff) { - bytes_.push_back(0xc5); - write_be16_value(static_cast(length)); - } else { - bytes_.push_back(0xc6); - write_be32_value(static_cast(length)); - } - bytes_.insert(bytes_.end(), value.begin(), value.end()); - } - - void write_array_len(uint32_t length) - { - if (length <= 15) { - bytes_.push_back(static_cast(0x90 | length)); - } else if (length <= 0xffff) { - bytes_.push_back(0xdc); - write_be16_value(static_cast(length)); - } else { - bytes_.push_back(0xdd); - write_be32_value(length); - } - } - - void write_map_len(uint32_t length) - { - if (length <= 15) { - bytes_.push_back(static_cast(0x80 | length)); - } else if (length <= 0xffff) { - bytes_.push_back(0xde); - write_be16_value(static_cast(length)); - } else { - bytes_.push_back(0xdf); - write_be32_value(length); - } - } - -private: - std::vector bytes_; - - void write_be16_value(uint16_t value) - { - bytes_.push_back(static_cast((value >> 8) & 0xff)); - bytes_.push_back(static_cast(value & 0xff)); - } - - void write_be32_value(uint32_t value) - { - bytes_.push_back(static_cast((value >> 24) & 0xff)); - bytes_.push_back(static_cast((value >> 16) & 0xff)); - bytes_.push_back(static_cast((value >> 8) & 0xff)); - bytes_.push_back(static_cast(value & 0xff)); - } - - void write_be64_value(uint64_t value) - { - write_be32_value(static_cast((value >> 32) & 0xffffffffull)); - write_be32_value(static_cast(value & 0xffffffffull)); - } -}; - -static std::string muninn_feedback_observed_at() -{ - const auto now = std::chrono::system_clock::now().time_since_epoch(); - const auto millis = std::chrono::duration_cast(now).count(); - return "unix-ms:" + std::to_string(millis); -} - -static void write_string_array(MediaMsgpackWriter &writer, const std::vector &values) -{ - writer.write_array_len(static_cast(values.size())); - for (const auto &value : values) { - writer.write_string(value); - } -} - -static void write_u64_array(MediaMsgpackWriter &writer, const std::vector &values) -{ - writer.write_array_len(static_cast(values.size())); - for (const auto value : values) { - writer.write_u64(value); - } -} - -static std::vector encode_muninn_receiver_feedback_payload( - const std::string &stream_id, - const std::string &session_id, - uint64_t frame_id, - const std::vector &missing_chunk_keys) -{ - MediaMsgpackWriter record; - record.write_array_len(11); - record.write_string(stream_id); - record.write_string(session_id); - record.write_string("starfire.obs"); - if (frame_id > 0) { - record.write_u64(frame_id - 1); - } else { - record.write_nil(); - } - write_u64_array(record, {}); - write_u64_array(record, {frame_id}); - record.write_bool(true); - record.write_i64(0); - record.write_i64(0); - record.write_string(muninn_feedback_observed_at()); - write_string_array(record, missing_chunk_keys); - - const std::string record_key = stream_id + ":" + session_id + ":feedback:starfire.obs"; - const std::string message_id = - "muninn-media:" + std::string(MuninnMediaReceiverFeedbackSchema) + ":" + record_key; - - MediaMsgpackWriter document; - document.write_map_len(9); - document.write_string("schemaId"); - document.write_string(MuninnMediaReceiverFeedbackSchema); - document.write_string("recordKey"); - document.write_string(record_key); - document.write_string("storedAt"); - document.write_string(muninn_feedback_observed_at()); - document.write_string("payloadEncoding"); - document.write_string("messagepack"); - document.write_string("payload"); - document.write_binary(record.bytes()); - document.write_string("sourceRuntimeId"); - document.write_string("starfire"); - document.write_string("sourceAgentId"); - document.write_nil(); - document.write_string("sourceRole"); - document.write_string("mimir.obs"); - document.write_string("tags"); - write_string_array(document, {"muninn.media"}); - - MediaMsgpackWriter wire; - wire.write_map_len(3); - wire.write_string("schemaVersion"); - wire.write_string("cultnet.document_put_raw.v0"); - wire.write_string("messageId"); - wire.write_string(message_id); - wire.write_string("document"); - wire.write_raw(document.bytes()); - return wire.bytes(); -} - static std::optional decode_muninn_media_payload(const uint8_t *payload, uint32_t payload_len) { try { @@ -1045,7 +823,7 @@ private: if (!last_remote_ || assembly.stream_id.empty() || assembly.session_id.empty()) { return; } - const auto payload = encode_muninn_receiver_feedback_payload( + const auto payload = muninn_media_wire::encode_receiver_feedback_payload( assembly.stream_id, assembly.session_id, assembly.frame_id, missing_chunk_keys); send_media_packet(payload, bridge_sequence++, *last_remote_); } diff --git a/native/obs_stem_source/muninn_media_wire.h b/native/obs_stem_source/muninn_media_wire.h new file mode 100644 index 0000000..a66753c --- /dev/null +++ b/native/obs_stem_source/muninn_media_wire.h @@ -0,0 +1,235 @@ +#pragma once + +#include +#include +#include +#include +#include + +namespace muninn_media_wire { + +inline constexpr const char *ReceiverFeedbackSchema = "muninn.media_receiver_feedback.v1"; + +class MsgpackWriter { +public: + const std::vector &bytes() const { return bytes_; } + + void write_raw(const std::vector &value) + { + bytes_.insert(bytes_.end(), value.begin(), value.end()); + } + + void write_nil() { bytes_.push_back(0xc0); } + + void write_bool(bool value) { bytes_.push_back(value ? 0xc3 : 0xc2); } + + void write_u64(uint64_t value) + { + if (value <= 0x7f) { + bytes_.push_back(static_cast(value)); + } else if (value <= 0xff) { + bytes_.push_back(0xcc); + bytes_.push_back(static_cast(value)); + } else if (value <= 0xffff) { + bytes_.push_back(0xcd); + write_be16_value(static_cast(value)); + } else if (value <= 0xffffffffull) { + bytes_.push_back(0xce); + write_be32_value(static_cast(value)); + } else { + bytes_.push_back(0xcf); + write_be64_value(value); + } + } + + void write_i64(int64_t value) + { + if (value >= 0) { + write_u64(static_cast(value)); + return; + } + if (value >= -32) { + bytes_.push_back(static_cast(0xe0 | (value + 32))); + } else if (value >= INT8_MIN) { + bytes_.push_back(0xd0); + bytes_.push_back(static_cast(value)); + } else if (value >= INT16_MIN) { + bytes_.push_back(0xd1); + write_be16_value(static_cast(value)); + } else if (value >= INT32_MIN) { + bytes_.push_back(0xd2); + write_be32_value(static_cast(value)); + } else { + bytes_.push_back(0xd3); + write_be64_value(static_cast(value)); + } + } + + void write_string(const std::string &value) + { + const auto length = value.size(); + if (length <= 31) { + bytes_.push_back(static_cast(0xa0 | length)); + } else if (length <= 0xff) { + bytes_.push_back(0xd9); + bytes_.push_back(static_cast(length)); + } else if (length <= 0xffff) { + bytes_.push_back(0xda); + write_be16_value(static_cast(length)); + } else { + bytes_.push_back(0xdb); + write_be32_value(static_cast(length)); + } + bytes_.insert(bytes_.end(), value.begin(), value.end()); + } + + void write_binary(const std::vector &value) + { + const auto length = value.size(); + if (length <= 0xff) { + bytes_.push_back(0xc4); + bytes_.push_back(static_cast(length)); + } else if (length <= 0xffff) { + bytes_.push_back(0xc5); + write_be16_value(static_cast(length)); + } else { + bytes_.push_back(0xc6); + write_be32_value(static_cast(length)); + } + bytes_.insert(bytes_.end(), value.begin(), value.end()); + } + + void write_array_len(uint32_t length) + { + if (length <= 15) { + bytes_.push_back(static_cast(0x90 | length)); + } else if (length <= 0xffff) { + bytes_.push_back(0xdc); + write_be16_value(static_cast(length)); + } else { + bytes_.push_back(0xdd); + write_be32_value(length); + } + } + + void write_map_len(uint32_t length) + { + if (length <= 15) { + bytes_.push_back(static_cast(0x80 | length)); + } else if (length <= 0xffff) { + bytes_.push_back(0xde); + write_be16_value(static_cast(length)); + } else { + bytes_.push_back(0xdf); + write_be32_value(length); + } + } + +private: + std::vector bytes_; + + void write_be16_value(uint16_t value) + { + bytes_.push_back(static_cast((value >> 8) & 0xff)); + bytes_.push_back(static_cast(value & 0xff)); + } + + void write_be32_value(uint32_t value) + { + bytes_.push_back(static_cast((value >> 24) & 0xff)); + bytes_.push_back(static_cast((value >> 16) & 0xff)); + bytes_.push_back(static_cast((value >> 8) & 0xff)); + bytes_.push_back(static_cast(value & 0xff)); + } + + void write_be64_value(uint64_t value) + { + write_be32_value(static_cast((value >> 32) & 0xffffffffull)); + write_be32_value(static_cast(value & 0xffffffffull)); + } +}; + +inline std::string feedback_observed_at() +{ + const auto now = std::chrono::system_clock::now().time_since_epoch(); + const auto millis = std::chrono::duration_cast(now).count(); + return "unix-ms:" + std::to_string(millis); +} + +inline void write_string_array(MsgpackWriter &writer, const std::vector &values) +{ + writer.write_array_len(static_cast(values.size())); + for (const auto &value : values) { + writer.write_string(value); + } +} + +inline void write_u64_array(MsgpackWriter &writer, const std::vector &values) +{ + writer.write_array_len(static_cast(values.size())); + for (const auto value : values) { + writer.write_u64(value); + } +} + +inline std::vector encode_receiver_feedback_payload( + const std::string &stream_id, + const std::string &session_id, + uint64_t frame_id, + const std::vector &missing_chunk_keys) +{ + MsgpackWriter record; + record.write_array_len(11); + record.write_string(stream_id); + record.write_string(session_id); + record.write_string("starfire.obs"); + if (frame_id > 0) { + record.write_u64(frame_id - 1); + } else { + record.write_nil(); + } + write_u64_array(record, {}); + write_u64_array(record, {frame_id}); + record.write_bool(true); + record.write_i64(0); + record.write_i64(0); + record.write_string(feedback_observed_at()); + write_string_array(record, missing_chunk_keys); + + const std::string record_key = stream_id + ":" + session_id + ":feedback:starfire.obs"; + const std::string message_id = + "muninn-media:" + std::string(ReceiverFeedbackSchema) + ":" + record_key; + + MsgpackWriter document; + document.write_map_len(9); + document.write_string("schemaId"); + document.write_string(ReceiverFeedbackSchema); + document.write_string("recordKey"); + document.write_string(record_key); + document.write_string("storedAt"); + document.write_string(feedback_observed_at()); + document.write_string("payloadEncoding"); + document.write_string("messagepack"); + document.write_string("payload"); + document.write_binary(record.bytes()); + document.write_string("sourceRuntimeId"); + document.write_string("starfire"); + document.write_string("sourceAgentId"); + document.write_nil(); + document.write_string("sourceRole"); + document.write_string("mimir.obs"); + document.write_string("tags"); + write_string_array(document, {"muninn.media"}); + + MsgpackWriter wire; + wire.write_map_len(3); + wire.write_string("schemaVersion"); + wire.write_string("cultnet.document_put_raw.v0"); + wire.write_string("messageId"); + wire.write_string(message_id); + wire.write_string("document"); + wire.write_raw(document.bytes()); + return wire.bytes(); +} + +} // namespace muninn_media_wire diff --git a/tools/muninn_media_decoder_probe.cpp b/tools/muninn_media_decoder_probe.cpp index 87a7cae..f48f052 100644 --- a/tools/muninn_media_decoder_probe.cpp +++ b/tools/muninn_media_decoder_probe.cpp @@ -7,14 +7,19 @@ #include #include +#include "muninn_media_wire.h" + struct MuninnMediaPayload { - enum class Kind { Unknown, Video, Audio }; + enum class Kind { Unknown, Video, Audio, Feedback }; Kind kind = Kind::Unknown; std::string stream_id; std::string session_id; uint64_t frame_id = 0; uint16_t chunk_index = 0; uint16_t chunk_count = 1; + uint64_t receiver_late_frames = 0; + uint64_t receiver_missing_chunks = 0; + bool receiver_requested_keyframe = false; std::vector payload; }; @@ -98,6 +103,14 @@ public: throw std::runtime_error("expected MessagePack unsigned integer"); } + bool read_bool() + { + const uint8_t tag = read_u8(); + if (tag == 0xc2) return false; + if (tag == 0xc3) return true; + throw std::runtime_error("expected MessagePack bool"); + } + void skip() { const uint8_t tag = read_u8(); @@ -271,6 +284,25 @@ static std::optional decode_muninn_media_payload(const uint8 media.payload = record.read_bytes(); return media; } + if (schema_id == "muninn.media_receiver_feedback.v1") { + if (record_len < 11) return std::nullopt; + MuninnMediaPayload media; + media.kind = MuninnMediaPayload::Kind::Feedback; + media.stream_id = record.read_string(); + media.session_id = record.read_string(); + record.skip(); // receiver_id + record.skip(); // highest_decodable_frame_id + record.skip(); // missing_frame_ids + media.receiver_late_frames = record.read_array_len(); + for (uint32_t index = 0; index < media.receiver_late_frames; ++index) record.skip(); + media.receiver_requested_keyframe = record.read_bool(); + record.skip(); // jitter_us + record.skip(); // decode_queue_us + record.skip(); // observed_at + media.receiver_missing_chunks = record.read_array_len(); + for (uint32_t index = 0; index < media.receiver_missing_chunks; ++index) record.skip(); + return media; + } return std::nullopt; } @@ -312,6 +344,14 @@ static std::optional> assemble_video_payload( static void print_payload(const MuninnMediaPayload &decoded) { + if (decoded.kind == MuninnMediaPayload::Kind::Feedback) { + std::cout << "feedback stream_id=" << decoded.stream_id + << " session_id=" << decoded.session_id + << " requested_keyframe=" << (decoded.receiver_requested_keyframe ? 1 : 0) + << " late_frames=" << decoded.receiver_late_frames + << " missing_chunks=" << decoded.receiver_missing_chunks << "\n"; + return; + } std::cout << (decoded.kind == MuninnMediaPayload::Kind::Video ? "video" : "audio"); if (decoded.kind == MuninnMediaPayload::Kind::Video) { std::cout << " frame_id=" << decoded.frame_id @@ -327,15 +367,22 @@ static void print_payload(const MuninnMediaPayload &decoded) int main(int argc, char **argv) { if (argc < 2) { - std::cerr << "usage: muninn_media_decoder_probe [more-wire.bin ...]\n"; + std::cerr << "usage: muninn_media_decoder_probe [more-wire.bin ...]\n" + << " muninn_media_decoder_probe --feedback-fixture\n"; return 2; } std::map video_assemblies; bool decoded_any = false; for (int index = 1; index < argc; ++index) { - std::ifstream file(argv[index], std::ios::binary); - std::vector bytes((std::istreambuf_iterator(file)), std::istreambuf_iterator()); + std::vector bytes; + if (std::string(argv[index]) == "--feedback-fixture") { + bytes = muninn_media_wire::encode_receiver_feedback_payload( + "muninn.raven.av.rudp", "raven:test:video", 42, {"42:1", "42:3"}); + } else { + std::ifstream file(argv[index], std::ios::binary); + bytes.assign(std::istreambuf_iterator(file), std::istreambuf_iterator()); + } const auto decoded = decode_muninn_media_payload(bytes.data(), static_cast(bytes.size())); if (!decoded) { std::cout << "none path=" << argv[index] << "\n";