diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index ef1383523ec..a704a8c2c11 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -6126,7 +6126,6 @@ abstract class MoqBroadcastProducerInterface { void close(); MoqBroadcastConsumer consume(); MoqBroadcastDynamic dynamic_(); - void finish(); MoqMediaProducer publishAudio({required MoqAudioInit init}); MoqMediaProducer publishAudioOnTrack({ required MoqTrackRequest request, @@ -6296,15 +6295,6 @@ class MoqBroadcastProducer implements MoqBroadcastProducerInterface { ); } - void finish() { - return rustCall((status) { - uniffi_moq_ffi_fn_method_moqbroadcastproducer_finish( - uniffiClonePointer(), - status, - ); - }, moqExceptionErrorHandler); - } - MoqMediaProducer publishAudio({required MoqAudioInit init}) { return rustCallWithLifter( (status) => uniffi_moq_ffi_fn_method_moqbroadcastproducer_publish_audio( @@ -10572,14 +10562,6 @@ external Pointer uniffi_moq_ffi_fn_method_moqbroadcastproducer_dynamic( Pointer uniffiStatus, ); -@Native, Pointer)>( - assetId: _uniffiAssetId, -) -external void uniffi_moq_ffi_fn_method_moqbroadcastproducer_finish( - Pointer ptr, - Pointer uniffiStatus, -); - @Native< Pointer Function(Pointer, RustBuffer, Pointer) >(assetId: _uniffiAssetId) @@ -12245,9 +12227,6 @@ external int uniffi_moq_ffi_checksum_method_moqbroadcastproducer_consume(); @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqbroadcastproducer_dynamic(); -@Native(assetId: _uniffiAssetId) -external int uniffi_moq_ffi_checksum_method_moqbroadcastproducer_finish(); - @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqbroadcastproducer_publish_audio(); @@ -12807,7 +12786,7 @@ void _checkApiChecksums() { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqoriginproducer_create_broadcast() != - 48971) { + 45756) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqoriginproducer_dynamic() != 56233) { @@ -12848,9 +12827,6 @@ void _checkApiChecksums() { if (uniffi_moq_ffi_checksum_method_moqbroadcastproducer_dynamic() != 55635) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } - if (uniffi_moq_ffi_checksum_method_moqbroadcastproducer_finish() != 29562) { - throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); - } if (uniffi_moq_ffi_checksum_method_moqbroadcastproducer_publish_audio() != 31691) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); diff --git a/doc/lib/c/index.md b/doc/lib/c/index.md index 6cac01a9517..ad5702c4718 100644 --- a/doc/lib/c/index.md +++ b/doc/lib/c/index.md @@ -47,7 +47,7 @@ beside `pkgconfig/`. - **Server.** `moq_server_listen` binds before it returns (a bad address or certificate fails there) and hands each incoming session to `on_request` as a request handle. Read `moq_session_request_path` and `_query` to route and authenticate, then `moq_session_request_accept` (a session handle, with origins like `moq_session_connect`) or `moq_session_request_reject` with an HTTP-style code (401 and 403 become the protocol's unauthorized close). An accepted session reports `1` once SETUP completes and never reconnects. `moq_server_addr` reports an ephemeral port and `moq_server_fingerprints` the hashes a client pins for a `tls_generate` certificate. `moq_server_close` stops listening; its terminal callback fires once the sockets are released. - **Demand.** A watcher on a published track (`moq_publish_track_demand`, `moq_publish_media_demand`, `moq_encode_video_demand`, `moq_encode_audio_demand`) calls `on_demand` with `MOQ_DEMAND_USED` or `MOQ_DEMAND_UNUSED` right away and again on every change, so an encoder on a battery-powered device runs only while someone is watching. The first call is the current state, so a track that went unused before the watcher existed still reports it. `moq_publish_demand_cancel` stops it; the terminal callback still fires. A container has no single demand and is refused. Demand follows the last real subscriber: an origin that served the track drops its source copy on the unused edge, so a cache linger never delays it. Only a relay keeps what it already delivered warm for 30 seconds, and a returning subscriber is served that cache only once the publisher confirms it is still current. - **Requests.** `moq_publish_dynamic` serves subscriptions to tracks the broadcast never declared: each arrives as a request handle, read its name with `moq_track_request_name`, then `moq_track_request_accept` (a raw track handle), `moq_track_request_video` / `_audio` (the media handle `moq_publish_video` / `_audio` return), or `moq_track_request_abort` with an application code the subscriber sees. Without a live handler an unknown name is refused. `moq_publish_track_dynamic` does the same for fetches of groups a track no longer has cached, delivered as `moq_group_request_*` (`sequence`, `priority`, `frame_start`); `moq_group_request_accept` starts the producer at `frame_start` so written frames keep their group indices. Register it with `moq_track_request_dynamic` before accepting a track that was itself requested by a fetch, so that pending group survives the transition. Both handlers stop with `moq_publish_dynamic_cancel`. -- **Everything the bindings can do** ([list](/lib/#what-every-binding-can-do)): media publish and consume with the catalog managed for you, raw pixels and PCM with the codec inside (`moq_encode_video`, `moq_encode_audio`, and the `moq_decode_*` mirrors), raw tracks with timestamps and datagrams, JSON and binary data tracks (snapshot or stream, each advertised in the catalog for as long as it lives), group fetch, catalog sections, shared video properties, and stalled hints. The three advertising operations are `moq_origin_create_broadcast` (unannounced producer, invisible to everyone), `moq_publish_announce` / `moq_publish_unannounce` (exact-path advertisement), `moq_publish_close` (ends the broadcast for good and releases its handle; `moq_publish_finish` is its deprecated alias), and `moq_origin_dynamic` (a claim over a path prefix and everything beneath it; `""` for everything). A route is a capability, not an inventory. `moq_origin_announced` takes a `moq_announce_config` (NULL or zeroed for everything): a literal prefix and an optional relative pattern filter; `moq_announce_update.prefix` stays relative to the origin, while `captures` reports what each wildcard matched when `has_captures` is true. Paths with a `.`-prefixed segment below the prefix are [hidden](/concept/moq-lite#hidden-broadcasts); set `hidden` or name the dot segment in `prefix` to list them. +- **Everything the bindings can do** ([list](/lib/#what-every-binding-can-do)): media publish and consume with the catalog managed for you, raw pixels and PCM with the codec inside (`moq_encode_video`, `moq_encode_audio`, and the `moq_decode_*` mirrors), raw tracks with timestamps and datagrams, JSON and binary data tracks (snapshot or stream, each advertised in the catalog for as long as it lives), group fetch, catalog sections, shared video properties, and stalled hints. The three advertising operations are `moq_origin_create_broadcast` (unannounced producer, invisible to everyone), `moq_publish_announce` / `moq_publish_unannounce` (exact-path advertisement), `moq_publish_close` (ends the broadcast for good and releases its handle), and `moq_origin_dynamic` (a claim over a path prefix and everything beneath it; `""` for everything). A route is a capability, not an inventory. `moq_origin_announced` takes a `moq_announce_config` (NULL or zeroed for everything): a literal prefix and an optional relative pattern filter; `moq_announce_update.prefix` stays relative to the origin, while `captures` reports what each wildcard matched when `has_captures` is true. Paths with a `.`-prefixed segment below the prefix are [hidden](/concept/moq-lite#hidden-broadcasts); set `hidden` or name the dot segment in `prefix` to list them. ```c moq_client_config config; diff --git a/doc/lib/dart/index.md b/doc/lib/dart/index.md index ff1a77178f4..0431f0fdff6 100644 --- a/doc/lib/dart/index.md +++ b/doc/lib/dart/index.md @@ -66,7 +66,7 @@ The three advertising operations: `moq.createBroadcast(path)` (or `origin.createBroadcast`) returns an unannounced producer, invisible to everyone; `broadcast.announce(route:)` / `broadcast.unannounce()` own that exact-path advertisement, and `broadcast.close()` ends the broadcast for good (a second -call is a no-op; `finish()` is its deprecated alias); `origin.dynamic_(prefix:, route:)` claims `prefix` and +call is a no-op); `origin.dynamic_(prefix:, route:)` claims `prefix` and every path beneath it (`''` for everything; Dart spells the origin method `dynamic_` because `dynamic` is reserved). Hold the returned handle while the claim should stay advertised, and reject the requests you will not serve. A diff --git a/doc/lib/go/index.md b/doc/lib/go/index.md index 227a50f9892..16115bc7b30 100644 --- a/doc/lib/go/index.md +++ b/doc/lib/go/index.md @@ -77,7 +77,7 @@ The three advertising operations: `client.CreateBroadcast(path)` (or `origin.CreateBroadcast`) returns an unannounced producer, invisible to everyone; `broadcast.Announce(route)` / `broadcast.Unannounce()` own that exact-path advertisement, and `broadcast.Close()` ends the broadcast for good (a second -call is a no-op; `Finish` is its deprecated alias); `origin.Dynamic(prefix, route)` claims `prefix` and every +call is a no-op); `origin.Dynamic(prefix, route)` claims `prefix` and every path beneath it (`""` for everything). Hold the returned `OriginDynamic` while the claim should stay advertised, and reject the requests you will not serve. A route is a capability, not an inventory. `Announced(options)` combines diff --git a/doc/lib/kt/index.md b/doc/lib/kt/index.md index 013a2d8489b..8a9f2c01140 100644 --- a/doc/lib/kt/index.md +++ b/doc/lib/kt/index.md @@ -58,8 +58,9 @@ Call `media.discontinuity()` when the source seeks, pauses, or changes its time The three advertising operations: `moq.createBroadcast(path)` (or `origin.createBroadcast`) returns an unannounced producer, invisible to everyone; `broadcast.announce(route)` / `broadcast.unannounce()` own that exact-path -advertisement, and `broadcast.close()` (or `use { }`) releases the producer, -ending the broadcast once no `dynamic()` handle remains; `origin.dynamic(prefix, route)` claims `prefix` and every +advertisement, and `broadcast.end()` ends the broadcast for good (a second call +is a no-op; Kotlin spells it `end` because `close()`, or `use { }`, releases the +handle, which ends the broadcast only once no `dynamic()` handle remains); `origin.dynamic(prefix, route)` claims `prefix` and every path beneath it (`""` for everything). Hold the returned `OriginDynamic` while the claim should stay advertised, and reject the requests you will not serve. A route is a capability, not an inventory. `announcements(config)` takes diff --git a/doc/lib/py/index.md b/doc/lib/py/index.md index 3a562d03a72..32e57bfc686 100644 --- a/doc/lib/py/index.md +++ b/doc/lib/py/index.md @@ -77,8 +77,7 @@ The three advertising operations, as the other bindings spell them: `client.create_broadcast(path)` (or `OriginProducer.create_broadcast`) returns an unannounced producer, invisible to everyone; `broadcast.announce(route)` / `broadcast.unannounce()` own that exact-path advertisement, and -`broadcast.close()` ends the broadcast for good (a second call is a no-op; -`finish()` is its deprecated alias); +`broadcast.close()` ends the broadcast for good (a second call is a no-op); `origin.dynamic(prefix, route)` claims `prefix` and every path beneath it (`""` for everything). Hold the returned handle while the claim should stay advertised, and reject the requests you will not serve. A route is a diff --git a/doc/lib/swift/index.md b/doc/lib/swift/index.md index 6867e3347d2..f757baaa0c6 100644 --- a/doc/lib/swift/index.md +++ b/doc/lib/swift/index.md @@ -62,8 +62,7 @@ Call `audio.discontinuity()` when the source seeks, pauses, or changes its time The three advertising operations: `session.publish.createBroadcast(path:)` returns an unannounced producer, invisible to everyone; `broadcast.announce(route:)` / `broadcast.unannounce()` own that exact-path advertisement, and -`broadcast.close()` ends the broadcast for good (a second call is a no-op; -`finish()` is its deprecated alias); +`broadcast.close()` ends the broadcast for good (a second call is a no-op); `session.publish.dynamic(prefix:route:)` claims `prefix` and every path beneath it (`""` for everything). Hold the returned `OriginDynamic` while the claim should stay advertised, and reject the requests you will not serve. A diff --git a/go/wrapper/publish.go b/go/wrapper/publish.go index 5b668b3e433..99fcbedc3d7 100644 --- a/go/wrapper/publish.go +++ b/go/wrapper/publish.go @@ -268,13 +268,6 @@ func (b *BroadcastProducer) Close() error { return b.inner.Close() } -// Finish ends the broadcast. -// -// Deprecated: use [BroadcastProducer.Close]; a broadcast end carries no cause. -func (b *BroadcastProducer) Finish() error { - return b.inner.Close() -} - // BroadcastDynamic is a stream of subscriber-requested tracks. type BroadcastDynamic struct { inner *ffi.MoqBroadcastDynamic diff --git a/js/net/src/broadcast.ts b/js/net/src/broadcast.ts index b744b042bb3..99ce9725776 100644 --- a/js/net/src/broadcast.ts +++ b/js/net/src/broadcast.ts @@ -25,7 +25,7 @@ let attachAnnouncer: (producer: Producer, announcer: Announcer) => void; class BroadcastState { requested = new Signal([]); pending = new Set(); - closed = new Once(); + closed = new Once(); tracks = new Map(); sequences = new Map(); // Live consumer handles sharing this state (see {@link Consumer.clone}). The broadcast @@ -45,10 +45,10 @@ function dequeueRequest(state: BroadcastState): track.Request | undefined { // // Once.set throws on a second settle, and the producer and each consumer handle close // independently, so this has to be idempotent. -function closeState(state: BroadcastState, abort?: Error) { +function closeState(state: BroadcastState) { if (state.closed.peek() !== undefined) return; - state.closed.set(abort ?? null); - for (const request of state.pending) request.reject(abort); + state.closed.set(null); + for (const request of state.pending) request.reject(); state.requested.mutate((requests) => { requests.length = 0; }); @@ -68,7 +68,7 @@ function subscribe( register = false, ): track.Subscriber { if (state.closed.peek() !== undefined) { - throw new Error(`broadcast is closed: ${state.closed.peek()}`); + throw new Error("broadcast is closed"); } const existing = state.tracks.get(name); @@ -103,7 +103,7 @@ async function resolveTrackInfo(state: BroadcastState, name: string): Promise { + get closed(): GetPromise { return this.#state.closed; } @@ -188,9 +188,7 @@ export class Producer { const request = dequeueRequest(this.#state); if (request) return request; - const closed = this.#state.closed.peek(); - if (closed instanceof Error) throw closed; - if (closed !== undefined) return undefined; + if (this.#state.closed.peek() !== undefined) return undefined; await Signal.race(this.#state.requested, this.#state.closed); } @@ -199,7 +197,7 @@ export class Producer { /** Insert a track that is served directly, without an on-demand request round-trip. */ insertTrack(track: track.Producer): void { if (this.#state.closed.peek() !== undefined) { - throw new Error(`broadcast is closed: ${this.#state.closed.peek()}`); + throw new Error("broadcast is closed"); } const existing = this.#state.tracks.get(track.name); @@ -252,7 +250,7 @@ export class Producer { */ announce(route: Route | { hops?: Route["hops"]; cost?: Route["cost"] | bigint } = Route.default): void { if (this.#state.closed.peek() !== undefined) { - throw new Error(`broadcast is closed: ${this.#state.closed.peek()}`); + throw new Error("broadcast is closed"); } if (!this.#announcer) throw new Error("broadcast is not attached to an origin"); this.#announcer.announce(Route.normalize(route)); @@ -267,13 +265,10 @@ export class Producer { } /** End the broadcast for good: retract it, serve no new tracks, and refuse a later {@link announce}. Idempotent. */ - close(): void; - /** @deprecated A broadcast end carries no cause; call `close()` without one. */ - close(abort?: Error): void; - close(abort?: Error) { + close(): void { this.#announcer?.unannounce(); this.#announcer = undefined; - closeState(this.#state, abort); + closeState(this.#state); } } @@ -312,13 +307,13 @@ export class Consumer { } /** - * Settles once the broadcast closes: `null` on a clean close, or the abort {@link Error}. + * Settles with `null` once the broadcast closes; a broadcast end carries no cause. * Peek it synchronously (`undefined` while open), observe it reactively, or `await` it. * * Shared by every {@link clone}: it settles once the last handle closes. The subscribing * wire layer peeks it to evict a closed entry from its per-path consume cache. */ - get closed(): GetPromise { + get closed(): GetPromise { return this.#state.closed; } @@ -350,9 +345,7 @@ export class Consumer { const request = dequeueRequest(this.#state); if (request) return request; - const closed = this.#state.closed.peek(); - if (closed instanceof Error) throw closed; - if (closed !== undefined) return undefined; + if (this.#state.closed.peek() !== undefined) return undefined; await Signal.race(this.#state.requested, this.#state.closed); } @@ -362,13 +355,10 @@ export class Consumer { * Release this handle. The broadcast is closed once this was the last live handle; * while other {@link clone}s remain open it stays live. */ - close(): void; - /** @deprecated A broadcast end carries no cause; call `close()` without one. */ - close(abort?: Error): void; - close(abort?: Error) { + close(): void { if (this.#closed) return; this.#closed = true; if (--this.#state.consumers > 0) return; - closeState(this.#state, abort); + closeState(this.#state); } } diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Moq.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Moq.kt index dec946bc2f8..c7d1f79f226 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Moq.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Moq.kt @@ -22,8 +22,8 @@ class Moq internal constructor( /** * Create an unannounced broadcast at [path], invisible to everyone until announced. * - * Advertise it with `announce` after populating tracks. `close()` (or `use`) ends it once no - * `dynamic()` handle remains. + * Advertise it with `announce` after populating tracks. `end()` ends it for good; `close()` + * (or `use`) releases the handle, which ends it once no `dynamic()` handle remains. */ fun createBroadcast(path: String): BroadcastProducer = session.publish().createBroadcast(path) diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Server.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Server.kt index 9257ba51ddf..1de4c55100c 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Server.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Server.kt @@ -33,8 +33,9 @@ class Server internal constructor( * Create a live broadcast at [path], served to incoming sessions. * * The origin announces the path so subscribers can discover it, becoming visible - * Advertise it with `announce` after populating tracks. `close()` (or `use`) - * ends it once no `dynamic()` handle remains. + * Advertise it with `announce` after populating tracks. `end()` ends it for + * good; `close()` (or `use`) releases the handle, which ends it once no + * `dynamic()` handle remains. */ fun createBroadcast(path: String): BroadcastProducer { val origin = publishOrigin ?: throw IllegalStateException("no publish origin configured") diff --git a/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt b/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt index 19efd52640d..07580cf0cc7 100644 --- a/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt +++ b/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt @@ -142,6 +142,19 @@ class SmokeTest { assertFailsWith { consumer.subscribeTrack("events", null) } } + @Test + fun `ending a broadcast ends it while a dynamic handle remains`() = runTest { + BroadcastProducer().use { broadcast -> + broadcast.dynamic().use { + val consumer = broadcast.consume() + // Releasing the producer alone would leave the dynamic handle holding it open. + broadcast.end() + broadcast.end() + assertFailsWith { consumer.subscribeTrack("events", null) } + } + } + } + @Test fun `broadcast updates shared video properties`() { BroadcastProducer().use { broadcast -> diff --git a/py/moq-rs/README.md b/py/moq-rs/README.md index 0902031a5ae..adcb20c59c1 100644 --- a/py/moq-rs/README.md +++ b/py/moq-rs/README.md @@ -155,7 +155,7 @@ client = moq.Client( - `.publish_video(format, init=b"", *, label=None, hint=None, track=None) → MediaProducer`. `init` may be empty for a format that resolves in band; a `VideoHint` pins catalog fields the stream can't reveal (bitrate) or publishes the catalog before the first keyframe. `track` names the track as in `publish_audio`. - `.encode_video(input, output, *, bandwidth=None) → VideoProducer`. Encode raw `VideoFrame`s inside the binding; `.write(frame)` each one. - `.encode_audio(name, input, output, *, bandwidth=None) → AudioProducer`. Encode raw PCM `AudioFrame`s; the codec is `output.codec`, e.g. `AudioCodec.opus()`, with `output.frame_duration_us` setting the Opus frame length. - - `.close()` ends the broadcast for good; a second call is a no-op. `.finish()` is its deprecated alias. + - `.close()` ends the broadcast for good; a second call is a no-op. - **`BroadcastDynamic`**. Async source of tracks requested by subscribers. - `await .requested_track() → TrackRequest`. Call `.accept()` on it for a `TrackProducer`, or `.abort(code)` to reject. - Async iterator yielding `TrackRequest` diff --git a/py/moq-rs/moq/publish.py b/py/moq-rs/moq/publish.py index 322009e6a0a..8138961e296 100644 --- a/py/moq-rs/moq/publish.py +++ b/py/moq-rs/moq/publish.py @@ -3,7 +3,6 @@ from __future__ import annotations import json -import warnings from typing import TYPE_CHECKING, Any from moq_ffi import ( @@ -850,8 +849,3 @@ def close(self) -> None: Tracks already subscribed carry on to their own end. Closing again is a no-op. """ self._inner.close() - - def finish(self) -> None: - """Deprecated: use :meth:`close`. A broadcast end carries no cause.""" - warnings.warn("use close(); a broadcast end carries no cause", DeprecationWarning, stacklevel=2) - self._inner.close() diff --git a/py/moq-rs/tests/test_local.py b/py/moq-rs/tests/test_local.py index 0c5238471bb..2ade484d309 100644 --- a/py/moq-rs/tests/test_local.py +++ b/py/moq-rs/tests/test_local.py @@ -315,12 +315,6 @@ def test_close_twice_is_a_noop(): broadcast.publish_audio(moq.AudioFormat.OPUS, opus_head()) -def test_finish_is_deprecated(): - broadcast = moq.BroadcastProducer() - with pytest.deprecated_call(): - broadcast.finish() - - async def test_announced_broadcast(): origin = moq.OriginProducer() _broadcast = create_announced(origin, "test/broadcast") diff --git a/quest/m1/README.md b/quest/m1/README.md index 5f913062187..fd5463199c6 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -27,8 +27,6 @@ transport, benchmark tooling); worktrees isolate commits, not semantics. - [Binding audio delay](/quest/m1/binding-surface.md) - moq-ffi, libmoq, and every wrapper configure and observe audio playout delay - [FFI shape](/quest/m1/ffi-shape/README.md) - the bindings mirror Rust's layers: net at the root, then media, json, audio, and video namespaces built from the handle below - [Track demand](/quest/m1/track-demand.md) - Rust and JS watch a track's subscribers through `demand()` alone -- [Kotlin end](/quest/m1/kotlin-end.md) - Kotlin exposes `close()` as `end()`, since `AutoCloseable.close()` takes the name -- [Remove finish](/quest/m1/broadcast-remove.md) - on dev, the deprecated broadcast end APIs are gone and `closed()` carries no cause - [Session close](/quest/m1/session-close.md) - a graceful session end withdraws announces and waits one second for the ack - [Close codes](/quest/m1/close-codes.md) - a client sees the peer's application close code over WebSocket and raw QUIC, like WebTransport - [JS caught up](/quest/m1/js-announce-caught-up.md) - @moq/net's announce consumer says when the initial set has landed, like Rust diff --git a/quest/m1/broadcast-remove.md b/quest/m1/broadcast-remove.md deleted file mode 100644 index fd8898f0ae9..00000000000 --- a/quest/m1/broadcast-remove.md +++ /dev/null @@ -1,21 +0,0 @@ -# [S] Remove finish - -## Goal - -On `dev`, the deprecated broadcast end APIs are gone in every language, and -waiting for a broadcast to close says only that it closed. - -## Plan - -This is a published API break, so it targets `dev`. - -- Remove `broadcast::Producer::finish`, `abort`, and `Consumer::is_finished`, - and the `finished` and `abort` fields they read. -- `Consumer::closed()`, `Consumer::poll_closed`, and `Dynamic::closed()` return - `()` rather than an `Error` cause. -- Remove the deprecated binding `finish` methods, `moq_publish_finish`, and - JS's `close(abort)` parameter. - -## Required - -- [Kotlin end](/quest/m1/kotlin-end.md) - Kotlin keeps a forced end once `finish` is gone diff --git a/quest/m1/kotlin-end.md b/quest/m1/kotlin-end.md deleted file mode 100644 index 44516dd62cc..00000000000 --- a/quest/m1/kotlin-end.md +++ /dev/null @@ -1,14 +0,0 @@ -# [XS] Kotlin end - -## Goal - -Kotlin can force-end a broadcast after `finish` is removed. Its -`MoqBroadcastProducer` exposes the moq-ffi `close()` as `end()`, because a -generated `close()` would collide with `AutoCloseable.close()`. - -## Plan - -- Rename the method for Kotlin only in `rs/moq-ffi/uniffi.toml`; every other - binding keeps `close()`. Kotlin's `close()` still only releases the handle. -- Test that `end()` ends the broadcast while a `dynamic()` handle is still - held. Document it in `doc/lib/kt`. diff --git a/rs/libmoq/src/api.rs b/rs/libmoq/src/api.rs index 987d9de8d4e..0c3f6e2418b 100644 --- a/rs/libmoq/src/api.rs +++ b/rs/libmoq/src/api.rs @@ -1963,14 +1963,6 @@ pub extern "C" fn moq_publish_close(broadcast: u32) -> i32 { }) } -/// Deprecated: use [moq_publish_close]. A broadcast end carries no cause. -/// -/// Returns a zero on success, or a negative code on failure. -#[unsafe(no_mangle)] -pub extern "C" fn moq_publish_finish(broadcast: u32) -> i32 { - moq_publish_close(broadcast) -} - /// Publish one audio codec as a new media track. /// /// The track is named after the format (`0.opus`), so a subscriber finds it diff --git a/rs/libmoq/src/test.rs b/rs/libmoq/src/test.rs index 1f6935011cc..a4820f11306 100644 --- a/rs/libmoq/src/test.rs +++ b/rs/libmoq/src/test.rs @@ -1822,10 +1822,6 @@ fn publish_close_releases_the_handle() { let broadcast = publish_broadcast(origin, b"close/twice"); assert_eq!(moq_publish_close(broadcast), 0); assert!(moq_publish_close(broadcast) < 0, "a closed handle is released"); - - // The deprecated alias still ends a broadcast. - let broadcast = publish_broadcast(origin, b"close/finish"); - assert_eq!(moq_publish_finish(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -2119,8 +2115,8 @@ fn announced_hides_dot_paths_unless_asked() { } } - assert_eq!(moq_publish_finish(stats), 0); - assert_eq!(moq_publish_finish(cam), 0); + assert_eq!(moq_publish_close(stats), 0); + assert_eq!(moq_publish_close(cam), 0); assert_eq!(moq_origin_close(origin), 0); } diff --git a/rs/moq-ffi/src/origin.rs b/rs/moq-ffi/src/origin.rs index 6480e9a8afd..e20f8d20d89 100644 --- a/rs/moq-ffi/src/origin.rs +++ b/rs/moq-ffi/src/origin.rs @@ -314,9 +314,8 @@ impl MoqOriginProducer { /// tracks; an on-demand handler is [`Self::dynamic`]. Create, `dynamic()` if /// tracks are served on demand, populate, then announce. /// - /// [`MoqBroadcastProducer::finish`] unpublishes immediately. Dropping the producer - /// without finishing also unpublishes, but subscribers observe the end as a - /// failure rather than a deliberate one. + /// [`MoqBroadcastProducer::close`] ends it for good; dropping its last handle, + /// `dynamic()` included, does the same. pub fn create_broadcast(&self, path: String) -> Result, MoqError> { let _guard = crate::ffi::enter(); // Surfaces Error::Unauthorized (out of scope) via the MoqError::Protocol conversion. diff --git a/rs/moq-ffi/src/producer.rs b/rs/moq-ffi/src/producer.rs index 847baa8a560..499cd6c8e2e 100644 --- a/rs/moq-ffi/src/producer.rs +++ b/rs/moq-ffi/src/producer.rs @@ -489,11 +489,6 @@ impl MoqBroadcastProducer { state.catalog.finish()?; Ok(()) } - - /// Deprecated: use `close()`. A broadcast end carries no cause. - pub fn finish(&self) -> Result<(), MoqError> { - self.close() - } } // ---- Dynamic Broadcast Producer ---- diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index 19ec67a80ef..95944688ead 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -1805,7 +1805,7 @@ async fn finish_unpublishes() { } }) .await; - assert!(removed.is_ok(), "finish should unpublish the broadcast"); + assert!(removed.is_ok(), "close should unpublish the broadcast"); } #[tokio::test] @@ -2138,7 +2138,7 @@ async fn video_decode_frame_ownership() { native.cancel(); drop((portable, native)); video.finish().unwrap(); - broadcast.finish().unwrap(); + broadcast.close().unwrap(); assert_eq!((retained.width(), retained.height()), (320, 240)); assert_eq!( diff --git a/rs/moq-ffi/uniffi.toml b/rs/moq-ffi/uniffi.toml index ddf0c572fed..4dda0865a5a 100644 --- a/rs/moq-ffi/uniffi.toml +++ b/rs/moq-ffi/uniffi.toml @@ -1,5 +1,5 @@ # Kotlin objects are AutoCloseable, and `close()` there releases the handle, -# which ends the broadcast like dropping the last Rust producer. A generated -# `close()` would collide with it. -[bindings.kotlin] -exclude = ["MoqBroadcastProducer.close"] +# which ends the broadcast only once no `dynamic()` handle remains. The +# generated forced end is `end()` so it doesn't collide with it. +[bindings.kotlin.rename] +"MoqBroadcastProducer.close" = "end" diff --git a/rs/moq-net/src/model/broadcast.rs b/rs/moq-net/src/model/broadcast.rs index 3b7f90306e0..c4d39408c7b 100644 --- a/rs/moq-net/src/model/broadcast.rs +++ b/rs/moq-net/src/model/broadcast.rs @@ -94,16 +94,9 @@ struct BroadcastState { // joined across per-session tracks. `None` for an ordinary broadcast. spliced: Option, - // Set once the broadcast ends: `Producer::close()`, an abort, or the last - // producer-side handle dropping. Every lookup after it answers `Unroutable`. + // Set once the broadcast ends: `Producer::close()` or the last producer-side + // handle dropping. Every lookup after it answers `Unroutable`. closing: bool, - - // Set only by the deprecated `Producer::finish()`, for `Consumer::is_finished`. - finished: bool, - - // The error passed to `Producer::abort()`, reported by `Consumer::closed`. - // `None` for a finish or a dropped producer (reported as `Error::Dropped`). - abort: Option, } /// The spliced (route-fed) half of a broadcast: logical tracks that outlive any @@ -447,31 +440,6 @@ impl Producer { self.alive.close(); } - #[doc(hidden)] - #[deprecated(note = "use close(); a broadcast end carries no cause")] - pub fn finish(&self) { - self.alive.end(true); - } - - #[doc(hidden)] - #[deprecated(note = "use close(); a broadcast end carries no cause")] - pub fn abort(self, err: Error) -> Result<(), Error> { - { - let mut state = self.state.lock(); - if state.closing { - return Err(Error::Closed); - } - state.closing = true; - state.abort = Some(err.clone()); - // Same as a finish: an unserved name is answerable now, with the reason the - // broadcast ended. Published tracks keep their cache (no cascade). - state.reject_unserved(err); - } - let _ = self.alive.token.close(); - self.alive.retire(); - Ok(()) - } - /// Return true if this is the same broadcast instance. pub fn is_clone(&self, other: &Self) -> bool { self.state.same_channel(&other.state) @@ -487,8 +455,7 @@ struct Alive { token: kio::Producer<()>, state: kio::Shared, // The advertisement of the broadcast's exact path, owned here so it retracts - // with the broadcast: on close, abort, or the last producer-side handle - // dropping. `None` for a standalone broadcast. + // with the broadcast: on close or the last producer-side handle dropping. `None` for a standalone broadcast. announcer: kio::Lock>, } @@ -510,33 +477,22 @@ impl Alive { /// End the broadcast. See [`Producer::close`]. fn close(&self) { - self.end(false); - } - - /// End the broadcast, recording the deprecated `finished` flag in the same locked - /// transition that claims the end, so a racing `abort` can't win after it's set. - fn end(&self, finished: bool) { { let mut state = self.state.lock(); if std::mem::replace(&mut state.closing, true) { return; } - state.finished = finished; // A name that was reserved or queued but never served can't arrive now, // and `Consumer::track` answers `Unroutable` for one asked about after this // point. Say the same to whoever asked earlier. state.reject_unserved(Error::Unroutable); } let _ = self.token.close(); - self.retire(); - } - /// End the broadcast's advertising for good: retract the standing advertisement - /// and drop the announcer, so a later `announce` fails with `Closed`. - fn retire(&self) { + // Drop the announcer for good, so a later `announce` fails with `Closed`. Dropped + // outside the announcer lock: the entry's removal re-syncs the origin's cursors + // under the origin's own lock. let announcer = self.announcer.lock().take(); - // Dropped outside the announcer lock: the entry's removal re-syncs the - // origin's cursors under the origin's own lock. drop(announcer); } } @@ -692,16 +648,13 @@ impl Dynamic { } /// Block until the broadcast ends, by [`Producer::close`] or every producer dropping. - /// - /// Returns [`Error::Dropped`], or the error passed to the deprecated `abort`. - pub async fn closed(&self) -> Error { + pub async fn closed(&self) { kio::wait(|waiter| self.poll_closed(waiter)).await } /// Poll-based variant of [`Self::closed`]. - pub fn poll_closed(&self, waiter: &kio::Waiter) -> Poll { - ready!(self.alive.token.poll_closed(waiter)); - Poll::Ready(self.state.read().abort.clone().unwrap_or(Error::Dropped)) + pub fn poll_closed(&self, waiter: &kio::Waiter) -> Poll<()> { + self.alive.token.poll_closed(waiter) } /// Return true if this is the same broadcast instance. @@ -892,11 +845,8 @@ impl Consumer { } /// Block until the broadcast ends, by [`Producer::close`] or every producer dropping. - /// - /// Returns [`Error::Dropped`], or the error passed to the deprecated `abort`. - pub async fn closed(&self) -> Error { - self.alive.closed().await; - self.state.read().abort.clone().unwrap_or(Error::Dropped) + pub async fn closed(&self) { + self.alive.closed().await } /// Returns true once the broadcast has ended. @@ -911,12 +861,6 @@ impl Consumer { self.state.read().closing } - #[doc(hidden)] - #[deprecated(note = "a broadcast end carries no cause")] - pub fn is_finished(&self) -> bool { - self.state.read().finished - } - /// Register a [`kio::Waiter`] that fires when the broadcast closes. /// /// Returns [`Poll::Ready`] if already closed, otherwise [`Poll::Pending`] after @@ -1281,7 +1225,7 @@ mod test { let consumer = producer.consume(); producer.close(); - assert!(matches!(consumer.closed().await, Error::Dropped)); + consumer.closed().await; assert!(matches!(consumer.track("video"), Err(Error::Unroutable))); assert!(matches!(clone.consume().track("video"), Err(Error::Unroutable))); @@ -1295,27 +1239,10 @@ mod test { let producer = Info::new().produce(); let consumer = producer.consume(); drop(producer); - assert!(matches!(consumer.closed().await, Error::Dropped)); + consumer.closed().await; assert!(matches!(consumer.track("video"), Err(Error::Unroutable))); } - /// The deprecated end APIs keep their old causes until they are removed. - #[tokio::test] - #[allow(deprecated)] - async fn deprecated_end_causes() { - let producer = Info::new().produce(); - let consumer = producer.consume(); - producer.abort(Error::Timeout).unwrap(); - assert!(matches!(consumer.closed().await, Error::Timeout)); - assert!(!consumer.is_finished()); - - let producer = Info::new().produce(); - let consumer = producer.consume(); - producer.finish(); - assert!(matches!(consumer.closed().await, Error::Dropped)); - assert!(consumer.is_finished()); - } - #[tokio::test] async fn requests() { let mut producer = Info::new().produce().dynamic(); @@ -1502,25 +1429,6 @@ mod test { assert!(matches!(pending.await, Err(Error::Unroutable))); } - /// The deprecated abort says why the broadcast ended, and an unserved name resolves - /// with that reason. - #[tokio::test] - #[allow(deprecated)] - async fn abort_resolves_a_reserved_name_with_its_reason() { - let producer = Info::new().produce(); - let consumer = producer.consume(); - - let request = producer.reserve_track("track1").unwrap(); - let pending = subscribe_pending!(consumer, "track1"); - - producer.abort(Error::Cancel).unwrap(); - assert!(matches!(pending.await, Err(Error::Cancel))); - - let track = request.accept(None); - let mut subscriber = track.subscribe(None); - assert!(matches!(subscriber.recv_group().await, Err(Error::Cancel))); - } - /// A request still queued for a handler is the same parking case reached from the /// consumer side, so it ends the same way. #[tokio::test] diff --git a/rs/moq-net/tests/loom.rs b/rs/moq-net/tests/loom.rs index bc36d5036c0..4731533d4c7 100644 --- a/rs/moq-net/tests/loom.rs +++ b/rs/moq-net/tests/loom.rs @@ -221,7 +221,7 @@ fn concurrent_tracks_drain_a_shared_pool() { for handle in handles { handle.join().unwrap(); } - broadcast.finish(); + broadcast.close(); drop(broadcast); assert_eq!(pool.used(), 0, "the pool kept a charge after every group was dropped"); diff --git a/rs/moq-relay/tests/drills.rs b/rs/moq-relay/tests/drills.rs index f6d3d52dd6f..24fb2e9a665 100644 --- a/rs/moq-relay/tests/drills.rs +++ b/rs/moq-relay/tests/drills.rs @@ -452,10 +452,10 @@ async fn cancel_under_backpressure_releases_the_reader(lane: Lane) { // Resource release: everything that session was feeding is closed, rather // than left parked on a subscription nobody will ever serve again. - let err = tokio::time::timeout(TIMEOUT, cancelled.closed()) + tokio::time::timeout(TIMEOUT, cancelled.closed()) .await .expect("the cancelled subscriber's broadcast never closed"); - println!("resource released: the cancelled broadcast closed with {err}"); + println!("resource released: the cancelled broadcast closed"); // ...and the relay survived it: a fresh subscriber still gets served, off the // upstream subscription the cancel left in place. diff --git a/rs/moq-transcode/src/rung.rs b/rs/moq-transcode/src/rung.rs index 1fc217c8237..08812ec2471 100644 --- a/rs/moq-transcode/src/rung.rs +++ b/rs/moq-transcode/src/rung.rs @@ -183,9 +183,9 @@ async fn live(rung: &Rung, producer: &mut moq_net::track::Producer) -> Result { + () = rung.broadcast.closed() => { // The source went away while idle; end the rung with it. - producer.clone().abort(err)?; + producer.clone().abort(moq_net::Error::Dropped)?; return Ok(Ended::Closed); } () = retire.fired() => { diff --git a/swift/Sources/Moq/Broadcast.swift b/swift/Sources/Moq/Broadcast.swift index 13e42a4d020..528f3401e18 100644 --- a/swift/Sources/Moq/Broadcast.swift +++ b/swift/Sources/Moq/Broadcast.swift @@ -350,9 +350,4 @@ public final class BroadcastProducer: Sendable { public func close() throws { try ffi.close() } - - @available(*, deprecated, renamed: "close", message: "A broadcast end carries no cause.") - public func finish() throws { - try ffi.close() - } } diff --git a/test/interop/clients/go/main.go b/test/interop/clients/go/main.go index 12c5d844d40..e99e482112b 100644 --- a/test/interop/clients/go/main.go +++ b/test/interop/clients/go/main.go @@ -79,12 +79,12 @@ func publish(ctx context.Context, url, broadcast string) error { } defer client.Close() - // Hold the producer for the lifetime of the publish loop; Finish unpublishes. + // Hold the producer for the lifetime of the publish loop; Close unpublishes. producer, err := client.CreateBroadcast(broadcast) if err != nil { return err } - defer producer.Finish() + defer producer.Close() media, err := producer.PublishVideoStream(moq.VideoFormatAvc3) if err != nil { @@ -113,7 +113,7 @@ func publish(ctx context.Context, url, broadcast string) error { publishTone(toneCtx, audio) }() // Let the tone unwind before any Finish, so no write races a finished - // producer. Runs before the deferred producer.Finish, and is a no-op once + // producer. Runs before the deferred producer.Close, and is a no-op once // the happy path below has already stopped and joined it. defer func() { stopTone()