refactor(ffi)!: the bindings mirror Rust's layers

cpp/moq/include/moq/moq.hpp

@@ -56,10 +56,7 @@ using BroadcastConsumer = MoqBroadcastConsumer;
 using BroadcastDynamic = MoqBroadcastDynamic;
 using BroadcastProducer = MoqBroadcastProducer;
 using BroadcastRequest = MoqBroadcastRequest;
-using CatalogConsumer = MoqCatalogConsumer;
 using Client = MoqClient;
-using ContainerProducer = MoqContainerProducer;
-using ContainerStreamProducer = MoqContainerStreamProducer;
 using FlateSnapshotProducer = MoqFlateSnapshotProducer;
 using FlateStreamProducer = MoqFlateStreamProducer;
 using GroupConsumer = MoqGroupConsumer;
@@ -69,10 +66,14 @@ using JsonSnapshotConsumer = MoqJsonSnapshotConsumer;
 using JsonSnapshotProducer = MoqJsonSnapshotProducer;
 using JsonStreamConsumer = MoqJsonStreamConsumer;
 using JsonStreamProducer = MoqJsonStreamProducer;
-using MediaConsumer = MoqMediaConsumer;
-using MediaGroupConsumer = MoqMediaGroupConsumer;
-using MediaProducer = MoqMediaProducer;
-using MediaStreamProducer = MoqMediaStreamProducer;
+using MediaCatalogConsumer = MoqMediaCatalogConsumer;
+using MediaCatalogProducer = MoqMediaCatalogProducer;
+using MediaContainerConsumer = MoqMediaContainerConsumer;
+using MediaContainerGroupConsumer = MoqMediaContainerGroupConsumer;
+using MediaContainerProducer = MoqMediaContainerProducer;
+using MediaContainerStreamProducer = MoqMediaContainerStreamProducer;
+using MediaTrackProducer = MoqMediaTrackProducer;
+using MediaTrackStreamProducer = MoqMediaTrackStreamProducer;
 using OriginConsumer = MoqOriginConsumer;
 using OriginDynamic = MoqOriginDynamic;
 using OriginProducer = MoqOriginProducer;
@@ -97,6 +98,8 @@ using AudioFrame = MoqAudioFrame;
 using AudioInit = MoqAudioInit;
 using Backoff = MoqBackoff;
 using Catalog = MoqCatalog;
+using ClientConfig = MoqClientConfig;
+using ClientTls = MoqClientTls;
 using ConnectionStats = MoqConnectionStats;
 using ContainerInit = MoqContainerInit;
 using Datagram = MoqDatagram;
@@ -106,10 +109,15 @@ using FlateConfig = MoqFlateConfig;
 using Frame = MoqFrame;
 using JsonSnapshotConfig = MoqJsonSnapshotConfig;
 using JsonStreamConfig = MoqJsonStreamConfig;
+using MediaContainerConfig = MoqMediaContainerConfig;
+using MediaContainerGroupConfig = MoqMediaContainerGroupConfig;
 using MediaFrame = MoqMediaFrame;
 using OriginConfig = MoqOriginConfig;
 using ProtocolError = MoqProtocolError;
+using QuicConfig = MoqQuicConfig;
 using Route = MoqRoute;
+using ServerConfig = MoqServerConfig;
+using ServerTls = MoqServerTls;
 using Subscription = MoqSubscription;
 using TrackInfo = MoqTrackInfo;
 using Video = MoqVideo;
@@ -121,6 +129,7 @@ using VideoFrame = MoqVideoFrame;
 using VideoHint = MoqVideoHint;
 using VideoInit = MoqVideoInit;
 using VideoProperties = MoqVideoProperties;
+using WebSocketConfig = MoqWebSocketConfig;
 using AnnounceEvent = MoqAnnounceEvent;
 using AudioFormat = MoqAudioFormat;
 using AudioSampleFormat = MoqAudioSampleFormat;
@@ -129,6 +138,7 @@ using Container = MoqContainer;
 using ContainerFormat = MoqContainerFormat;
 using Error = MoqError;
 using ErrorScope = MoqErrorScope;
+using MediaTarget = MoqMediaTarget;
 using ProtocolKind = MoqProtocolKind;
 using Transport = MoqTransport;
 using VideoCodec = MoqVideoCodec;

cpp/moq/test/doc-samples.hpp

@@ -28,7 +28,7 @@ inline void ok(moq::expected<void> result) {
 }
 
 inline std::shared_ptr<moq::Session> session;
-inline std::shared_ptr<moq::MediaConsumer> media;
+inline std::shared_ptr<moq::MediaContainerConsumer> media;
 inline std::vector<uint8_t> opus_init;
 inline std::vector<uint8_t> packet;
 inline std::vector<uint8_t> rgba;

cpp/moq/test/probe.cpp

@@ -100,7 +100,7 @@ Coroutine read_one(std::shared_ptr<moq::TrackConsumer> consumer, std::optional<m
 
 int main() {
     // A synchronous error is a returned value, not an exception.
-    auto unbound = moq::Server::init();
+    auto unbound = ok(moq::Server::init({}), "server init");
     auto fingerprints = unbound->cert_fingerprints();
     CHECK(!fingerprints);
     CHECK(std::holds_alternative<moq::Error::kBind>(fingerprints.error().get_variant()));
@@ -113,21 +113,23 @@ int main() {
     auto track = ok(broadcast->publish_track("data", std::nullopt), "publish_track");
     ok(broadcast->announce({}), "announce");
 
-    auto server = moq::Server::init();
-    ok(server->set_bind("127.0.0.1:0"), "set_bind");
-    ok(server->set_tls_generate({"localhost"}), "set_tls_generate");
-    ok(server->set_publish(origin), "set_publish");
+    moq::ServerConfig server_config;
+    server_config.bind = "127.0.0.1:0";
+    server_config.tls.generate = {"localhost"};
+    server_config.publish = origin;
+    auto server = ok(moq::Server::init(server_config), "server init");
     auto addr = ok(server->listen().get(), "listen");
 
     // Both halves of the handshake are futures, so they run concurrently while this
     // thread blocks on one at a time.
-    auto client = moq::Client::init();
-    ok(client->set_tls_verify(false), "set_tls_verify");
+    moq::ClientConfig client_config;
+    client_config.tls.insecure = true;
+    auto client = ok(moq::Client::init(client_config), "client init");
     auto accepting = server->accept();
     auto connecting = client->connect("https://" + addr);
     auto request = ok(accepting.get(), "accept");
     CHECK(request != nullptr);
-    auto served = ok(request->accept().get(), "request accept");
+    auto served = ok(request->accept(nullptr, nullptr).get(), "request accept");
     auto session = ok(connecting.get(), "connect");
 
     // Subscriber: resolve the announced broadcast and subscribe to its track.
@@ -205,7 +207,7 @@ int main() {
     auto ended = consumer->read_frame().get();
     CHECK(ended && !ended->has_value());
 
-    auto closed = moq::Client::init();
+    auto closed = ok(moq::Client::init({}), "client init");
     closed->cancel();
     auto refused = closed->connect("https://" + addr).get();
     CHECK(!refused);

cpp/obs/src/moq-output.cpp

@@ -161,24 +161,6 @@ bool MoQOutput::Start()
 		return false;
 	}
 
-	auto next = std::make_shared<Attempt>();
-	next->client = moq::Client::init();
-	next->url = server;
-
-	// Advanced settings live on the service alongside the URL and path. With the group
-	// switched off the client keeps the library defaults.
-	OBSDataAutoRelease service_settings = obs_service_get_settings(service);
-	std::string invalid;
-	if (!MoQSettings::Configure(service_settings, *next->client, &invalid)) {
-		// Refusing to start beats connecting with a setting the user asked for
-		// quietly dropped.
-		LOG_ERROR("Invalid advanced MoQ settings: %s", invalid.c_str());
-		std::lock_guard<std::recursive_mutex> signal_lock(signal_mutex);
-		obs_output_set_last_error(output, ("Invalid advanced MoQ settings: " + invalid).c_str());
-		obs_output_signal_stop(output, OBS_OUTPUT_CONNECT_FAILED);
-		return false;
-	}
-
 	// Create the broadcast on an origin of our own, then announce it so subscribers
 	// can discover it. Stop() finishes it, so each Start creates a fresh one.
 	LOG_INFO("Publishing broadcast: %s", path.c_str());
@@ -192,11 +174,29 @@ bool MoQOutput::Start()
 		LOG_ERROR("Failed to announce broadcast: %s", announced.error().to_string().c_str());
 		return false;
 	}
-	if (auto wired = next->client->set_publish(next_origin); !wired) {
-		LOG_ERROR("Failed to publish through the client: %s", wired.error().to_string().c_str());
+
+	// Advanced settings live on the service alongside the URL and path. With the group
+	// switched off the client keeps the library defaults.
+	moq::ClientConfig config;
+	config.publish = next_origin;
+	OBSDataAutoRelease service_settings = obs_service_get_settings(service);
+	MoQSettings::Configure(service_settings, config);
+	auto client = moq::Client::init(config);
+	if (!client) {
+		// Refusing to start beats connecting with a setting the user asked for
+		// quietly dropped.
+		std::string invalid = client.error().to_string();
+		LOG_ERROR("Invalid advanced MoQ settings: %s", invalid.c_str());
+		std::lock_guard<std::recursive_mutex> signal_lock(signal_mutex);
+		obs_output_set_last_error(output, ("Invalid advanced MoQ settings: " + invalid).c_str());
+		obs_output_signal_stop(output, OBS_OUTPUT_CONNECT_FAILED);
 		return false;
 	}
 
+	auto next = std::make_shared<Attempt>();
+	next->client = *client;
+	next->url = server;
+
 	LOG_INFO("Connecting to MoQ server: %s", MoQRedactUrl(server).c_str());
 
 	// Held from the connect through obs_output_begin_data_capture. Every
@@ -654,7 +654,7 @@ void MoQOutput::VideoInit(obs_encoder_t *encoder)
 	hint.optimize_for_latency = true;
 	config.hint = hint;
 
-	auto track = broadcast->publish_video(config);
+	auto track = moq::MediaTrackProducer::video(broadcast, moq::MediaTarget::kNamed{}, config);
 	if (!track) {
 		LOG_ERROR("Failed to initialize video track: %s", track.error().to_string().c_str());
 		video_tracks[encoder] = nullptr;
@@ -715,7 +715,7 @@ void MoQOutput::AudioInit(obs_encoder_t *encoder)
 	if (extra_data && extra_size > 0)
 		config.data.assign(extra_data, extra_data + extra_size);
 
-	auto track = broadcast->publish_audio(config);
+	auto track = moq::MediaTrackProducer::audio(broadcast, moq::MediaTarget::kNamed{}, config);
 	if (!track) {
 		LOG_ERROR("Failed to initialize audio track: %s", track.error().to_string().c_str());
 		audio_tracks[encoder] = nullptr;

cpp/obs/src/moq-output.h

@@ -141,8 +141,8 @@ class MoQOutput {
 	std::shared_ptr<moq::OriginProducer> origin;
 	std::shared_ptr<moq::BroadcastProducer> broadcast;
 	// An encoder maps to null when its track failed to initialize, so it isn't retried.
-	std::map<obs_encoder_t *, std::shared_ptr<moq::MediaProducer>> video_tracks;
-	std::map<obs_encoder_t *, std::shared_ptr<moq::MediaProducer>> audio_tracks;
+	std::map<obs_encoder_t *, std::shared_ptr<moq::MediaTrackProducer>> video_tracks;
+	std::map<obs_encoder_t *, std::shared_ptr<moq::MediaTrackProducer>> audio_tracks;
 
 	// Retired sessions still draining their finished tracks, which the destructor
 	// waits out. Guarded by signal_mutex.

cpp/obs/src/moq-settings.cpp

@@ -28,8 +28,11 @@ constexpr const char *QUIC_MAX_STREAMS = "quic_max_streams";
 constexpr const char *WEBSOCKET_ENABLED = "websocket_enabled";
 constexpr const char *WEBSOCKET_DELAY = "websocket_delay_ms";
 
-// moq-ffi documents these defaults on the MoqClient setters but does not report them,
-// so they are repeated here. The backoff defaults come from moq::Backoff itself.
+// moq-ffi documents these defaults on MoqClientConfig but leaves each field null rather
+// than reporting it, so they are repeated here.
+constexpr long long BACKOFF_INITIAL_DEFAULT_MS = 1000;
+constexpr long long BACKOFF_MAX_DEFAULT_MS = 5000;
+constexpr long long BACKOFF_TIMEOUT_DEFAULT_MS = 10000;
 constexpr long long QUIC_MAX_STREAMS_DEFAULT = 1024;
 constexpr bool WEBSOCKET_ENABLED_DEFAULT = true;
 constexpr long long WEBSOCKET_DELAY_DEFAULT_MS = 200;
@@ -78,17 +81,11 @@ Field File(const char *key, const char *label, const char *filter, const char *t
 	return Field{key, label, tooltip, Kind::File, 0, 0, 0, false, 0, "", {}, false, filter};
 }
 
-long long Millis(uint64_t us)
-{
-	return static_cast<long long>(us / 1000);
-}
-
 } // namespace
 
 const std::vector<Field> &Fields()
 {
 	static const std::vector<Field> fields = [] {
-		const moq::Backoff backoff{};
 		std::vector<Field> f;
 
 		f.push_back(Text(BIND, "Bind address",
@@ -109,11 +106,11 @@ const std::vector<Field> &Fields()
 				 "Trust this CA instead of the system roots."));
 
 		// Reconnect.
-		f.push_back(Number(BACKOFF_INITIAL, "Reconnect delay (ms)", Millis(backoff.initial_us), 1, 60000, 100,
+		f.push_back(Number(BACKOFF_INITIAL, "Reconnect delay (ms)", BACKOFF_INITIAL_DEFAULT_MS, 1, 60000, 100,
 				   "Delay before the first reconnect attempt; it grows from here."));
-		f.push_back(Number(BACKOFF_MAX, "Reconnect delay cap (ms)", Millis(backoff.max_us), 1, 600000, 1000,
+		f.push_back(Number(BACKOFF_MAX, "Reconnect delay cap (ms)", BACKOFF_MAX_DEFAULT_MS, 1, 600000, 1000,
 				   "Ceiling on the growing reconnect delay."));
-		f.push_back(Number(BACKOFF_TIMEOUT, "Give up after (ms)", Millis(backoff.timeout_us), 0, 3600000, 1000,
+		f.push_back(Number(BACKOFF_TIMEOUT, "Give up after (ms)", BACKOFF_TIMEOUT_DEFAULT_MS, 0, 3600000, 1000,
 				   "Total time to keep retrying before the stream fails. Also how long the "
 				   "broadcast lingers for viewers across the gap. 0 retries forever."));
 
@@ -199,52 +196,29 @@ void AddProperties(obs_properties_t *props)
 	obs_properties_add_group(props, ENABLED, "Advanced", OBS_GROUP_CHECKABLE, group);
 }
 
-bool Configure(obs_data_t *settings, moq::Client &client, std::string *error)
+void Configure(obs_data_t *settings, moq::ClientConfig &config)
 {
 	if (!settings || !obs_data_get_bool(settings, ENABLED))
-		return true; // advanced off: the client keeps the library defaults
-
-	// Stops at the first setter moq-ffi rejects, naming the setting it came from.
-	auto apply = [&](const char *key, moq::expected<void> result) {
-		if (!result && error)
-			*error = std::string(key) + ": " + result.error().to_string();
-		return result.has_value();
-	};
-
-	if (const char *bind = OptionalString(settings, BIND)) {
-		if (!apply(BIND, client.set_bind(bind)))
-			return false;
-	}
+		return; // advanced off: the client keeps the library defaults
 
-	if (obs_data_get_bool(settings, TLS_DISABLE_VERIFY)) {
-		if (!apply(TLS_DISABLE_VERIFY, client.set_tls_verify(false)))
-			return false;
-	}
-	if (const char *fingerprint = OptionalString(settings, TLS_FINGERPRINT)) {
-		if (!apply(TLS_FINGERPRINT, client.set_tls_fingerprints({fingerprint})))
-			return false;
-	}
-	if (const char *root = OptionalString(settings, TLS_ROOT)) {
-		if (!apply(TLS_ROOT, client.set_tls_roots({root})))
-			return false;
-	}
+	if (const char *bind = OptionalString(settings, BIND))
+		config.bind = bind;
 
-	moq::Backoff backoff{};
-	backoff.initial_us = Amount(settings, BACKOFF_INITIAL) * 1000;
-	backoff.max_us = Amount(settings, BACKOFF_MAX) * 1000;
-	backoff.timeout_us = Amount(settings, BACKOFF_TIMEOUT) * 1000;
-	if (!apply(BACKOFF_INITIAL, client.set_backoff(backoff)))
-		return false;
+	if (obs_data_get_bool(settings, TLS_DISABLE_VERIFY))
+		config.tls.insecure = true;
+	if (const char *fingerprint = OptionalString(settings, TLS_FINGERPRINT))
+		config.tls.fingerprints = {fingerprint};
+	if (const char *root = OptionalString(settings, TLS_ROOT))
+		config.tls.roots = {root};
 
-	if (!apply(QUIC_MAX_STREAMS, client.set_quic_max_streams(Amount(settings, QUIC_MAX_STREAMS))))
-		return false;
+	config.backoff.initial_us = Amount(settings, BACKOFF_INITIAL) * 1000;
+	config.backoff.max_us = Amount(settings, BACKOFF_MAX) * 1000;
+	config.backoff.timeout_us = Amount(settings, BACKOFF_TIMEOUT) * 1000;
 
-	if (!apply(WEBSOCKET_ENABLED, client.set_websocket_enabled(obs_data_get_bool(settings, WEBSOCKET_ENABLED))))
-		return false;
-	if (!apply(WEBSOCKET_DELAY, client.set_websocket_delay(Amount(settings, WEBSOCKET_DELAY) * 1000)))
-		return false;
+	config.quic.max_streams = Amount(settings, QUIC_MAX_STREAMS);
 
-	return true;
+	config.websocket.enabled = obs_data_get_bool(settings, WEBSOCKET_ENABLED);
+	config.websocket.delay_us = Amount(settings, WEBSOCKET_DELAY) * 1000;
 }
 
 } // namespace MoQSettings

cpp/obs/src/moq-settings.h

@@ -12,7 +12,7 @@
 // One set of keys on an obs_data_t backs every surface that edits them: the service
 // properties page, the dock's advanced dialog, and the output that reads them at connect
 // time. The fields are described once in Fields() so both UIs are generated from the
-// same list; adding a knob means adding a Field entry and one setter call in Configure.
+// same list; adding a knob means adding a Field entry and one config field in Configure.
 namespace MoQSettings {
 
 // Whether any of this applies. With it off the output dials with the library defaults and
@@ -77,11 +77,12 @@ void Defaults(obs_data_t *settings);
 // Add the checkable "Advanced" group to a properties list.
 void AddProperties(obs_properties_t *props);
 
-// Apply these settings to a client before it connects.
+// Fill a client's config from these settings before the client is built.
 //
-// Returns false with the offending setting and moq-ffi's reason in `error` when a value
-// is rejected, such as a bind address that doesn't parse; the caller should refuse to
-// start rather than dial with a setting the user asked for silently dropped.
-bool Configure(obs_data_t *settings, moq::Client &client, std::string *error);
+// moq-ffi validates the record when it builds the client: a value it rejects, such as a
+// bind address that doesn't parse, fails `moq::Client::init` with a `Config` error naming
+// it, and the caller should refuse to start rather than dial with a setting the user asked
+// for silently dropped.
+void Configure(obs_data_t *settings, moq::ClientConfig &config);
 
 } // namespace MoQSettings