GameCult/Mimir
Watch
9
0
Fork
You've already forked Mimir
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Share Muninn media feedback wire encoder

Browse source
This commit is contained in:
Meta Vens 2026-06-18 21:13:48 +01:00
commit 2f0e0bdfcb
4 changed files with 295 additions and 229 deletions
Download patch file Download diff file Expand all files Collapse all files

6
native/obs_stem_source/CMakeLists.txt
View file

@ -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})

228
native/obs_stem_source/mimir_obs_muninn_source.cpp
View file

@ -6,10 +6,11 @@
#include <obs-module.h>
#include "muninn_media_wire.h"
#include <algorithm>
#include <atomic>
#include <chrono>
#include <climits>
#include <condition_variable>
#include <cstdint>
#include <cctype>
@ -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<uint8_t> &bytes() const { return bytes_; }
void write_raw(const std::vector<uint8_t> &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<uint8_t>(value));
} else if (value <= 0xff) {
bytes_.push_back(0xcc);
bytes_.push_back(static_cast<uint8_t>(value));
} else if (value <= 0xffff) {
bytes_.push_back(0xcd);
write_be16_value(static_cast<uint16_t>(value));
} else if (value <= 0xffffffffull) {
bytes_.push_back(0xce);
write_be32_value(static_cast<uint32_t>(value));
} else {
bytes_.push_back(0xcf);
write_be64_value(value);
}
}
void write_i64(int64_t value)
{
if (value >= 0) {
write_u64(static_cast<uint64_t>(value));
return;
}
if (value >= -32) {
bytes_.push_back(static_cast<uint8_t>(0xe0 | (value + 32)));
} else if (value >= INT8_MIN) {
bytes_.push_back(0xd0);
bytes_.push_back(static_cast<uint8_t>(value));
} else if (value >= INT16_MIN) {
bytes_.push_back(0xd1);
write_be16_value(static_cast<uint16_t>(value));
} else if (value >= INT32_MIN) {
bytes_.push_back(0xd2);
write_be32_value(static_cast<uint32_t>(value));
} else {
bytes_.push_back(0xd3);
write_be64_value(static_cast<uint64_t>(value));
}
}
void write_string(const std::string &value)
{
const auto length = value.size();
if (length <= 31) {
bytes_.push_back(static_cast<uint8_t>(0xa0 | length));
} else if (length <= 0xff) {
bytes_.push_back(0xd9);
bytes_.push_back(static_cast<uint8_t>(length));
} else if (length <= 0xffff) {
bytes_.push_back(0xda);
write_be16_value(static_cast<uint16_t>(length));
} else {
bytes_.push_back(0xdb);
write_be32_value(static_cast<uint32_t>(length));
}
bytes_.insert(bytes_.end(), value.begin(), value.end());
}
void write_binary(const std::vector<uint8_t> &value)
{
const auto length = value.size();
if (length <= 0xff) {
bytes_.push_back(0xc4);
bytes_.push_back(static_cast<uint8_t>(length));
} else if (length <= 0xffff) {
bytes_.push_back(0xc5);
write_be16_value(static_cast<uint16_t>(length));
} else {
bytes_.push_back(0xc6);
write_be32_value(static_cast<uint32_t>(length));
}
bytes_.insert(bytes_.end(), value.begin(), value.end());
}
void write_array_len(uint32_t length)
{
if (length <= 15) {
bytes_.push_back(static_cast<uint8_t>(0x90 | length));
} else if (length <= 0xffff) {
bytes_.push_back(0xdc);
write_be16_value(static_cast<uint16_t>(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<uint8_t>(0x80 | length));
} else if (length <= 0xffff) {
bytes_.push_back(0xde);
write_be16_value(static_cast<uint16_t>(length));
} else {
bytes_.push_back(0xdf);
write_be32_value(length);
}
}
private:
std::vector<uint8_t> bytes_;
void write_be16_value(uint16_t value)
{
bytes_.push_back(static_cast<uint8_t>((value >> 8) & 0xff));
bytes_.push_back(static_cast<uint8_t>(value & 0xff));
}
void write_be32_value(uint32_t value)
{
bytes_.push_back(static_cast<uint8_t>((value >> 24) & 0xff));
bytes_.push_back(static_cast<uint8_t>((value >> 16) & 0xff));
bytes_.push_back(static_cast<uint8_t>((value >> 8) & 0xff));
bytes_.push_back(static_cast<uint8_t>(value & 0xff));
}
void write_be64_value(uint64_t value)
{
write_be32_value(static_cast<uint32_t>((value >> 32) & 0xffffffffull));
write_be32_value(static_cast<uint32_t>(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<std::chrono::milliseconds>(now).count();
return "unix-ms:" + std::to_string(millis);
}
static void write_string_array(MediaMsgpackWriter &writer, const std::vector<std::string> &values)
{
writer.write_array_len(static_cast<uint32_t>(values.size()));
for (const auto &value : values) {
writer.write_string(value);
}
}
static void write_u64_array(MediaMsgpackWriter &writer, const std::vector<uint64_t> &values)
{
writer.write_array_len(static_cast<uint32_t>(values.size()));
for (const auto value : values) {
writer.write_u64(value);
}
}
static std::vector<uint8_t> encode_muninn_receiver_feedback_payload(
const std::string &stream_id,
const std::string &session_id,
uint64_t frame_id,
const std::vector<std::string> &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<MuninnMediaPayload> 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_);
}

235
native/obs_stem_source/muninn_media_wire.h Normal file
View file

@ -0,0 +1,235 @@
#pragma once
#include <chrono>
#include <climits>
#include <cstdint>
#include <string>
#include <vector>
namespace muninn_media_wire {
inline constexpr const char *ReceiverFeedbackSchema = "muninn.media_receiver_feedback.v1";
class MsgpackWriter {
public:
const std::vector<uint8_t> &bytes() const { return bytes_; }
void write_raw(const std::vector<uint8_t> &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<uint8_t>(value));
} else if (value <= 0xff) {
bytes_.push_back(0xcc);
bytes_.push_back(static_cast<uint8_t>(value));
} else if (value <= 0xffff) {
bytes_.push_back(0xcd);
write_be16_value(static_cast<uint16_t>(value));
} else if (value <= 0xffffffffull) {
bytes_.push_back(0xce);
write_be32_value(static_cast<uint32_t>(value));
} else {
bytes_.push_back(0xcf);
write_be64_value(value);
}
}
void write_i64(int64_t value)
{
if (value >= 0) {
write_u64(static_cast<uint64_t>(value));
return;
}
if (value >= -32) {
bytes_.push_back(static_cast<uint8_t>(0xe0 | (value + 32)));
} else if (value >= INT8_MIN) {
bytes_.push_back(0xd0);
bytes_.push_back(static_cast<uint8_t>(value));
} else if (value >= INT16_MIN) {
bytes_.push_back(0xd1);
write_be16_value(static_cast<uint16_t>(value));
} else if (value >= INT32_MIN) {
bytes_.push_back(0xd2);
write_be32_value(static_cast<uint32_t>(value));
} else {
bytes_.push_back(0xd3);
write_be64_value(static_cast<uint64_t>(value));
}
}
void write_string(const std::string &value)
{
const auto length = value.size();
if (length <= 31) {
bytes_.push_back(static_cast<uint8_t>(0xa0 | length));
} else if (length <= 0xff) {
bytes_.push_back(0xd9);
bytes_.push_back(static_cast<uint8_t>(length));
} else if (length <= 0xffff) {
bytes_.push_back(0xda);
write_be16_value(static_cast<uint16_t>(length));
} else {
bytes_.push_back(0xdb);
write_be32_value(static_cast<uint32_t>(length));
}
bytes_.insert(bytes_.end(), value.begin(), value.end());
}
void write_binary(const std::vector<uint8_t> &value)
{
const auto length = value.size();
if (length <= 0xff) {
bytes_.push_back(0xc4);
bytes_.push_back(static_cast<uint8_t>(length));
} else if (length <= 0xffff) {
bytes_.push_back(0xc5);
write_be16_value(static_cast<uint16_t>(length));
} else {
bytes_.push_back(0xc6);
write_be32_value(static_cast<uint32_t>(length));
}
bytes_.insert(bytes_.end(), value.begin(), value.end());
}
void write_array_len(uint32_t length)
{
if (length <= 15) {
bytes_.push_back(static_cast<uint8_t>(0x90 | length));
} else if (length <= 0xffff) {
bytes_.push_back(0xdc);
write_be16_value(static_cast<uint16_t>(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<uint8_t>(0x80 | length));
} else if (length <= 0xffff) {
bytes_.push_back(0xde);
write_be16_value(static_cast<uint16_t>(length));
} else {
bytes_.push_back(0xdf);
write_be32_value(length);
}
}
private:
std::vector<uint8_t> bytes_;
void write_be16_value(uint16_t value)
{
bytes_.push_back(static_cast<uint8_t>((value >> 8) & 0xff));
bytes_.push_back(static_cast<uint8_t>(value & 0xff));
}
void write_be32_value(uint32_t value)
{
bytes_.push_back(static_cast<uint8_t>((value >> 24) & 0xff));
bytes_.push_back(static_cast<uint8_t>((value >> 16) & 0xff));
bytes_.push_back(static_cast<uint8_t>((value >> 8) & 0xff));
bytes_.push_back(static_cast<uint8_t>(value & 0xff));
}
void write_be64_value(uint64_t value)
{
write_be32_value(static_cast<uint32_t>((value >> 32) & 0xffffffffull));
write_be32_value(static_cast<uint32_t>(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<std::chrono::milliseconds>(now).count();
return "unix-ms:" + std::to_string(millis);
}
inline void write_string_array(MsgpackWriter &writer, const std::vector<std::string> &values)
{
writer.write_array_len(static_cast<uint32_t>(values.size()));
for (const auto &value : values) {
writer.write_string(value);
}
}
inline void write_u64_array(MsgpackWriter &writer, const std::vector<uint64_t> &values)
{
writer.write_array_len(static_cast<uint32_t>(values.size()));
for (const auto value : values) {
writer.write_u64(value);
}
}
inline std::vector<uint8_t> encode_receiver_feedback_payload(
const std::string &stream_id,
const std::string &session_id,
uint64_t frame_id,
const std::vector<std::string> &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

55
tools/muninn_media_decoder_probe.cpp
View file

@ -7,14 +7,19 @@
#include <string>
#include <vector>
#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<uint8_t> 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<MuninnMediaPayload> 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<std::vector<uint8_t>> 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 <wire.bin> [more-wire.bin ...]\n";
std::cerr << "usage: muninn_media_decoder_probe <wire.bin> [more-wire.bin ...]\n"
<< " muninn_media_decoder_probe --feedback-fixture\n";
return 2;
}
std::map<std::string, MuninnVideoFrameAssembly> video_assemblies;
bool decoded_any = false;
for (int index = 1; index < argc; ++index) {
std::ifstream file(argv[index], std::ios::binary);
std::vector<uint8_t> bytes((std::istreambuf_iterator<char>(file)), std::istreambuf_iterator<char>());
std::vector<uint8_t> 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<char>(file), std::istreambuf_iterator<char>());
}
const auto decoded = decode_muninn_media_payload(bytes.data(), static_cast<uint32_t>(bytes.size()));
if (!decoded) {
std::cout << "none path=" << argv[index] << "\n";