diff --git a/cpp/obs/src/moq-output.cpp b/cpp/obs/src/moq-output.cpp index bbe39e6879..7a5e071951 100644 --- a/cpp/obs/src/moq-output.cpp +++ b/cpp/obs/src/moq-output.cpp @@ -301,10 +301,10 @@ void MoQOutput::Reset() } audio_tracks.clear(); - // Finish the broadcast so the origin unpublishes it immediately; Start() + // Close the broadcast so the origin retracts it immediately; Start() // creates a fresh one on restart. if (broadcast > 0) { - moq_publish_finish(broadcast); + moq_publish_close(broadcast); broadcast = 0; } } diff --git a/cpp/obs/test/moq-output-test.cpp b/cpp/obs/test/moq-output-test.cpp index e28039ea60..c20c9d30c8 100644 --- a/cpp/obs/test/moq-output-test.cpp +++ b/cpp/obs/test/moq-output-test.cpp @@ -233,7 +233,7 @@ int32_t moq_publish_announce(uint32_t, const moq_route *) return 0; } -int32_t moq_publish_finish(uint32_t) +int32_t moq_publish_close(uint32_t) { return 0; } diff --git a/dart/moq/test/moq_test.dart b/dart/moq/test/moq_test.dart index 8fcc824ab2..278456ee19 100644 --- a/dart/moq/test/moq_test.dart +++ b/dart/moq/test/moq_test.dart @@ -102,7 +102,8 @@ void main() { client.close(); serverSession.cancel(code: 0); track.finish(); - broadcast.finish(); + broadcast.close(); + broadcast.close(); // a second close is a no-op server.close(); }); diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index 4ada2c2a1c..82da0be25b 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -8,9 +8,7 @@ import "dart:ffi"; import "dart:io" show Platform, File, Directory; import "dart:isolate"; import "dart:typed_data"; - import "package:ffi/ffi.dart"; - import "uniffi_runtime.dart"; export "uniffi_runtime.dart"; @@ -6125,6 +6123,7 @@ abstract class MoqBroadcastProducerInterface { required MoqJsonStreamConfig config, }); void announce({required MoqRoute route}); + void close(); MoqBroadcastConsumer consume(); MoqBroadcastDynamic dynamic_(); void finish(); @@ -6266,6 +6265,15 @@ class MoqBroadcastProducer implements MoqBroadcastProducerInterface { }, moqExceptionErrorHandler); } + void close() { + return rustCall((status) { + uniffi_moq_ffi_fn_method_moqbroadcastproducer_close( + uniffiClonePointer(), + status, + ); + }, moqExceptionErrorHandler); + } + MoqBroadcastConsumer consume() { return rustCallWithLifter( (status) => uniffi_moq_ffi_fn_method_moqbroadcastproducer_consume( @@ -10530,6 +10538,14 @@ external void uniffi_moq_ffi_fn_method_moqbroadcastproducer_announce( Pointer uniffiStatus, ); +@Native, Pointer)>( + assetId: _uniffiAssetId, +) +external void uniffi_moq_ffi_fn_method_moqbroadcastproducer_close( + Pointer ptr, + Pointer uniffiStatus, +); + @Native Function(Pointer, Pointer)>( assetId: _uniffiAssetId, ) @@ -12202,6 +12218,9 @@ uniffi_moq_ffi_checksum_method_moqbroadcastproducer_publish_json_stream(); @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqbroadcastproducer_announce(); +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_method_moqbroadcastproducer_close(); + @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqbroadcastproducer_consume(); @@ -12727,7 +12746,7 @@ void _checkApiChecksums() { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqannouncedbroadcast_available() != - 42497) { + 37458) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqannouncedbroadcast_cancel() != 63175) { @@ -12799,13 +12818,16 @@ void _checkApiChecksums() { if (uniffi_moq_ffi_checksum_method_moqbroadcastproducer_announce() != 13700) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } + if (uniffi_moq_ffi_checksum_method_moqbroadcastproducer_close() != 19191) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } if (uniffi_moq_ffi_checksum_method_moqbroadcastproducer_consume() != 27634) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqbroadcastproducer_dynamic() != 55635) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } - if (uniffi_moq_ffi_checksum_method_moqbroadcastproducer_finish() != 7183) { + 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() != diff --git a/doc/lib/c/index.md b/doc/lib/c/index.md index 7a3c5b08dd..308710a8e3 100644 --- a/doc/lib/c/index.md +++ b/doc/lib/c/index.md @@ -42,7 +42,7 @@ and `target/include/moq.h`. - **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), 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 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); 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; `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 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); 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 564d4e06ed..d7d31ee23c 100644 --- a/doc/lib/dart/index.md +++ b/doc/lib/dart/index.md @@ -65,7 +65,8 @@ await for (final request in server.requests()) { 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; `origin.dynamic_(prefix:, route:)` claims `prefix` and +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 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 001f5a50e4..5b3ffe9e47 100644 --- a/doc/lib/go/index.md +++ b/doc/lib/go/index.md @@ -66,7 +66,7 @@ video, _ := broadcast.EncodeVideo( ) _ = video.Write(moq.VideoFrame{TimestampUs: pts, Data: rgba}) _ = broadcast.Announce(moq.Route{}) -broadcast.Finish() // keep the producer reachable while publishing, then finish explicitly +broadcast.Close() // keep the producer reachable while publishing, then close explicitly ``` For locally encoded media, call `MediaProducer.Flush(timestampUs)` after `WriteFrame` with the same broadcast-clock PTS. It measures catalog jitter at the transport handoff. File, pipe, and network imports should omit `Flush`; built-in encoders observe their own output. @@ -74,7 +74,8 @@ For locally encoded media, call `MediaProducer.Flush(timestampUs)` after `WriteF 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; `origin.Dynamic(prefix, route)` claims `prefix` and every +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 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 a241ad6b0c..2b8e9d5795 100644 --- a/doc/lib/kt/index.md +++ b/doc/lib/kt/index.md @@ -56,7 +56,8 @@ Moq.connect("https://relay.example.com").use { moq -> 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; `origin.dynamic(prefix, route)` claims `prefix` and every +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 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 a41cb7255b..10b2edd914 100644 --- a/doc/lib/py/index.md +++ b/doc/lib/py/index.md @@ -74,7 +74,9 @@ For already-encoded live output, call `audio.flush(timestamp_us)` after each `au 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; +`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 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/rs/moq-net.md b/doc/lib/rs/moq-net.md index 2f56c399b6..7c4bb1fd8b 100644 --- a/doc/lib/rs/moq-net.md +++ b/doc/lib/rs/moq-net.md @@ -113,8 +113,12 @@ Three operations, on an origin: - `broadcast.announce(route)` / `broadcast.unannounce()` own that advertisement. Announcing again re-prices the standing route, which competes on cost with remote routes at the same path (a tie goes to the local - broadcast). The route retracts on `unannounce()`, `finish()`, or the last + broadcast). The route retracts on `unannounce()`, `close()`, or the last producer dropping; tracks already in flight carry on to their own end. +- `broadcast.close()` ends the broadcast for good: it retracts, leaves local + discovery, and answers every later track lookup with `Unroutable`. Tracks + already subscribed carry on to their own end. It can never be announced + again. Dropping the last producer does the same. - `origin.dynamic(prefix, route)` claims `prefix` and every path beneath it (`""` claims everything). Hold the returned `origin::Dynamic` while the claim should stay advertised; drop it to retract. A request beneath it with diff --git a/doc/lib/swift/index.md b/doc/lib/swift/index.md index a9aa8b3c44..413b9d7554 100644 --- a/doc/lib/swift/index.md +++ b/doc/lib/swift/index.md @@ -59,7 +59,9 @@ For already-encoded live output, call `audio.flush(timestampUs:)` after `writeFr 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; +`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); `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/example_test.go b/go/wrapper/example_test.go index 84eef29901..60c3c40d29 100644 --- a/go/wrapper/example_test.go +++ b/go/wrapper/example_test.go @@ -53,8 +53,8 @@ func ExampleClient_CreateBroadcast() { if err != nil { log.Fatal(err) } - // Finishing unpublishes the broadcast immediately. - defer broadcast.Finish() + // Closing ends the broadcast for good. + defer broadcast.Close() media, err := broadcast.PublishAudio(moq.AudioFormatOpus, opusHead()) if err != nil { @@ -91,7 +91,7 @@ func ExampleBroadcastProducer_PublishVideo_videoHint() { if err != nil { log.Fatal(err) } - defer broadcast.Finish() + defer broadcast.Close() media, err := broadcast.PublishVideo(moq.VideoFormatAvc3, nil, moq.WithVideoHint(moq.VideoHint{})) if err != nil { diff --git a/go/wrapper/moq_test.go b/go/wrapper/moq_test.go index 0a91ad7927..936fd7e5df 100644 --- a/go/wrapper/moq_test.go +++ b/go/wrapper/moq_test.go @@ -115,7 +115,7 @@ func TestDynamicBroadcastRequest(t *testing.T) { if err := track.Finish(); err != nil { t.Fatal(err) } - if err := served.Finish(); err != nil { + if err := served.Close(); err != nil { t.Fatal(err) } } @@ -135,11 +135,27 @@ func TestPublishAudioLifecycle(t *testing.T) { if err := media.Finish(); err != nil { t.Fatal(err) } - if err := broadcast.Finish(); err != nil { + if err := broadcast.Close(); err != nil { t.Fatal(err) } } +func TestBroadcastCloseTwiceIsNoop(t *testing.T) { + broadcast, err := moq.NewBroadcastProducer() + if err != nil { + t.Fatal(err) + } + if err := broadcast.Close(); err != nil { + t.Fatal(err) + } + if err := broadcast.Close(); err != nil { + t.Fatalf("second Close: %v", err) + } + if _, err := broadcast.PublishTrack("events", nil); err == nil { + t.Fatal("PublishTrack after Close succeeded") + } +} + func TestEncodeAudioWithOpusObject(t *testing.T) { // The producer retains the codec, so releasing the codec and the // config in either order must still encode. @@ -179,7 +195,7 @@ func TestEncodeAudioWithOpusObject(t *testing.T) { if err := producer.Finish(); err != nil { t.Fatal(err) } - if err := broadcast.Finish(); err != nil { + if err := broadcast.Close(); err != nil { t.Fatal(err) } } @@ -221,7 +237,7 @@ func TestEncodeAudioFrameDurations(t *testing.T) { t.Fatalf("err = %v, want ErrAudio: 2 ms is not an opus frame duration", err) } - if err := broadcast.Finish(); err != nil { + if err := broadcast.Close(); err != nil { t.Fatal(err) } } @@ -235,7 +251,7 @@ func TestVideoPropertiesUseDefaultedFields(t *testing.T) { if err := broadcast.SetVideoProperties(moq.VideoProperties{Rotation: &rotation}); err != nil { t.Fatal(err) } - if err := broadcast.Finish(); err != nil { + if err := broadcast.Close(); err != nil { t.Fatal(err) } } @@ -350,7 +366,7 @@ func TestDecodeVideoFormat(t *testing.T) { if err := video.Finish(); err != nil { t.Fatal(err) } - if err := broadcast.Finish(); err != nil { + if err := broadcast.Close(); err != nil { t.Fatal(err) } } @@ -806,7 +822,7 @@ func TestDynamicTrackRequest(t *testing.T) { if err != nil { t.Fatal(err) } - defer broadcast.Finish() + defer broadcast.Close() dynamic, err := broadcast.Dynamic() if err != nil { @@ -882,7 +898,7 @@ func TestDynamicTrackRequestCanPublishAudio(t *testing.T) { if err != nil { t.Fatal(err) } - defer broadcast.Finish() + defer broadcast.Close() dynamic, err := broadcast.Dynamic() if err != nil { @@ -974,7 +990,7 @@ func TestRecvGroupCancelRace(t *testing.T) { if err != nil { t.Fatal(err) } - defer broadcast.Finish() + defer broadcast.Close() var wg sync.WaitGroup for i := 0; i < 16; i++ { @@ -1007,7 +1023,7 @@ func TestConsumerCancelConcurrent(t *testing.T) { if err != nil { t.Fatal(err) } - defer broadcast.Finish() + defer broadcast.Close() track, err := broadcast.PublishTrack("x", nil) if err != nil { @@ -1075,7 +1091,7 @@ func TestRequestBroadcastCancelKeepsTheOrigin(t *testing.T) { if err != nil { t.Fatal(err) } - defer served.Finish() + defer served.Close() resolved := make(chan error, 1) go func() { @@ -1108,7 +1124,7 @@ func TestSubscribeTrackCancelKeepsTheBroadcast(t *testing.T) { if err != nil { t.Fatal(err) } - defer broadcast.Finish() + defer broadcast.Close() dynamic, err := broadcast.Dynamic() if err != nil { @@ -1177,7 +1193,7 @@ func TestUsedCancelKeepsTheTrack(t *testing.T) { if err != nil { t.Fatal(err) } - defer broadcast.Finish() + defer broadcast.Close() track, err := broadcast.PublishTrack("status", nil) if err != nil { @@ -1224,7 +1240,7 @@ func TestCancelDoesNotLeakGoroutines(t *testing.T) { if err != nil { t.Fatal(err) } - defer broadcast.Finish() + defer broadcast.Close() dynamic, err := broadcast.Dynamic() if err != nil { @@ -1338,7 +1354,7 @@ func TestBroadcastIsReachableOnlyWhileAnnounced(t *testing.T) { if _, err := consumer.RequestBroadcast(ctx, "live"); err != nil { t.Fatal(err) } - if err := broadcast.Finish(); err != nil { + if err := broadcast.Close(); err != nil { t.Fatal(err) } } diff --git a/go/wrapper/origin.go b/go/wrapper/origin.go index 67eb77e040..78642147c1 100644 --- a/go/wrapper/origin.go +++ b/go/wrapper/origin.go @@ -49,9 +49,8 @@ func (o *OriginProducer) Dynamic(prefix string, route Route) (*OriginDynamic, er // // The broadcast is invisible and unroutable, for this origin's consumers and // peers alike, until [BroadcastProducer.Announce]. Announce it after -// populating tracks. Finish unpublishes immediately, while dropping the -// producer without finishing also unpublishes but reads to subscribers as a -// failure rather than a deliberate end. +// populating tracks. [BroadcastProducer.Close] ends it for good; dropping the +// last handle does the same. func (o *OriginProducer) CreateBroadcast(path string) (*BroadcastProducer, error) { inner, err := o.inner.CreateBroadcast(path) if err != nil { diff --git a/go/wrapper/publish.go b/go/wrapper/publish.go index 6eccc6d8f6..27c7ff2087 100644 --- a/go/wrapper/publish.go +++ b/go/wrapper/publish.go @@ -262,9 +262,17 @@ func (b *BroadcastProducer) RemoveCatalogSection(name string) error { return b.inner.RemoveCatalogSection(name) } -// Finish closes the broadcast. +// Close ends the broadcast for good: it retracts and serves no new tracks. +// Tracks already subscribed carry on to their own end. Closing again is a no-op. +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.Finish() + return b.inner.Close() } // BroadcastDynamic is a stream of subscriber-requested tracks. diff --git a/go/wrapper/reconnect_test.go b/go/wrapper/reconnect_test.go index 72e71536d2..762a46c38d 100644 --- a/go/wrapper/reconnect_test.go +++ b/go/wrapper/reconnect_test.go @@ -166,7 +166,7 @@ func TestReconnectAcrossRelayRestart(t *testing.T) { if err != nil { t.Fatal(err) } - defer broadcast.Finish() + defer broadcast.Close() track, err := broadcast.PublishTrack("data", nil) if err != nil { diff --git a/js/net/src/broadcast.ts b/js/net/src/broadcast.ts index 41cbeca2ef..17f706cc75 100644 --- a/js/net/src/broadcast.ts +++ b/js/net/src/broadcast.ts @@ -265,7 +265,10 @@ export class Producer { this.#announcer?.unannounce(); } - /** Close the broadcast, optionally with an error to abort waiters. Idempotent. */ + /** 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) { this.#announcer?.unannounce(); this.#announcer = undefined; @@ -355,9 +358,12 @@ export class Consumer { } /** - * Release this handle. The broadcast is closed (optionally with an error to abort waiters) - * once this was the last live handle; while other {@link clone}s remain open it stays live. + * 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) { if (this.#closed) return; this.#closed = true; diff --git a/js/net/src/origin.test.ts b/js/net/src/origin.test.ts index 47d044158e..7609952773 100644 --- a/js/net/src/origin.test.ts +++ b/js/net/src/origin.test.ts @@ -271,8 +271,9 @@ test("closing the origin closes every routed broadcast", async () => { expect(origin.closed.peek()).toBe(abort); expect(consumer.closed.peek()).toBe(abort); - expect(a.closed.peek()).toBe(abort); - expect(b.closed.peek()).toBe(abort); + // A broadcast end carries no cause, so the routed broadcasts close cleanly. + expect(a.closed.peek()).toBeNull(); + expect(b.closed.peek()).toBeNull(); expect(wireOf(consumer).routes(Path.from("a"))).toBe(false); expect(() => publish(origin, Path.from("late"))).toThrow(); diff --git a/js/net/src/origin.ts b/js/net/src/origin.ts index 28dd06e62d..82386b8a04 100644 --- a/js/net/src/origin.ts +++ b/js/net/src/origin.ts @@ -174,7 +174,7 @@ class ServeState { finishRequest(request, err); } for (const [path, front] of this.served) { - front.close(abort); + front.close(); this.onChange(path); } this.served.clear(); @@ -788,7 +788,7 @@ export class Producer implements Table { if (this.#state.closed.peek() !== undefined) return; this.#state.closed.set(abort ?? null); for (const front of this.#state.created?.values() ?? []) { - front.close(abort); + front.close(); } this.#state.created = undefined; this.#state.local.update(() => undefined); diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Moq.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Moq.kt index 728c381c8e..dec946bc2f 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Moq.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Moq.kt @@ -22,7 +22,8 @@ class Moq internal constructor( /** * Create an unannounced broadcast at [path], invisible to everyone until announced. * - * Advertise it with `announce` after populating tracks. `finish()` unpublishes immediately. + * Advertise it with `announce` after populating tracks. `close()` (or `use`) 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 e085291a69..9257ba51dd 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Server.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Server.kt @@ -33,8 +33,8 @@ 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. `finish()` - * unpublishes immediately. + * Advertise it with `announce` after populating tracks. `close()` (or `use`) + * 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 048ee430c2..9cd8e32517 100644 --- a/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt +++ b/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt @@ -132,6 +132,16 @@ class SmokeTest { assertEquals(ConnectionStatus.CONNECTED, status) } + @Test + fun `closing a broadcast ends it, and closing again is a no-op`() = runTest { + val broadcast = BroadcastProducer() + val consumer = broadcast.consume() + // AutoCloseable.close() releases the last producer handle, which ends the broadcast. + broadcast.close() + broadcast.close() + assertFailsWith { consumer.subscribeTrack("events", null) } + } + @Test fun `broadcast updates shared video properties`() { BroadcastProducer().use { broadcast -> @@ -302,7 +312,6 @@ class SmokeTest { server.createBroadcast("live").use { broadcast -> broadcast.announce(Route()) broadcast.unannounce() - broadcast.finish() } } } diff --git a/py/moq-rs/README.md b/py/moq-rs/README.md index 6add3e7adb..0902031a5a 100644 --- a/py/moq-rs/README.md +++ b/py/moq-rs/README.md @@ -68,7 +68,7 @@ async def main(): # Clean up audio.finish() - broadcast.finish() + broadcast.close() asyncio.run(main()) @@ -132,7 +132,7 @@ client = moq.Client( - **`Server(bind="[::]:443", *, tls_cert=(), tls_key=(), tls_generate=(), publish=None, subscribe=None)`**. Async context manager + async iterator of incoming `Request`s. - `.local_addr`. The bound address (useful when binding to port `0`). - `.cert_fingerprints()`. SHA-256 fingerprints of the configured TLS certificates, for `serverCertificateHashes` browser cert pinning. - - `.create_broadcast(path) → BroadcastProducer`. Create an unannounced broadcast, invisible to everyone; `announce()` makes it discoverable and reachable; `finish()` unpublishes it. + - `.create_broadcast(path) → BroadcastProducer`. Create an unannounced broadcast, invisible to everyone; `announce()` makes it discoverable and reachable; `close()` ends it. - **`Request`**. An incoming session, yielded by `async for request in server`. - `.url`, `.path`, `.query`, `.transport`. The query-free path is uniform across transports; the root or missing path is `""`. The encoded query may contain credentials. - `.set_publish(origin)`, `.set_consume(origin)`. Per-request overrides, captured at `accept()`. Raise if the request is already answered, cancelled, or currently accepting. @@ -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. - - `.finish()` + - `.close()` ends the broadcast for good; a second call is a no-op. `.finish()` is its deprecated alias. - **`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/origin.py b/py/moq-rs/moq/origin.py index f4a7c1886b..517ebe1210 100644 --- a/py/moq-rs/moq/origin.py +++ b/py/moq-rs/moq/origin.py @@ -248,8 +248,7 @@ def create_broadcast(self, path: str) -> BroadcastProducer: and peers alike, until :meth:`BroadcastProducer.announce`. Announce it after populating tracks. Create, :meth:`dynamic` if tracks are served on demand, populate, then announce. - ``finish()`` unpublishes immediately, while dropping the producer without - finishing also unpublishes but reads to subscribers as a failure rather - than a deliberate end. + :meth:`BroadcastProducer.close` ends it for good; dropping the last + handle does the same. """ return BroadcastProducer._from_inner(self._inner.create_broadcast(path)) diff --git a/py/moq-rs/moq/publish.py b/py/moq-rs/moq/publish.py index cbaf29ba13..ae44fb4ad6 100644 --- a/py/moq-rs/moq/publish.py +++ b/py/moq-rs/moq/publish.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import warnings from typing import TYPE_CHECKING, Any from moq_ffi import ( @@ -839,6 +840,14 @@ def consume(self) -> BroadcastConsumer: return BroadcastConsumer(self._inner.consume()) + def close(self) -> None: + """End the broadcast for good: retract it and serve no new tracks. + + Tracks already subscribed carry on to their own end. Closing again is a no-op. + """ + self._inner.close() + def finish(self) -> None: - """Finish the broadcast, closing its tracks and unpublishing it.""" - self._inner.finish() + """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 ad004a566f..d8b23fb367 100644 --- a/py/moq-rs/tests/test_local.py +++ b/py/moq-rs/tests/test_local.py @@ -127,7 +127,7 @@ def test_publish_media_lifecycle(): media = broadcast.publish_audio(moq.AudioFormat.OPUS, opus_head()) media.write_frame(b"opus frame", 1000) media.finish() - broadcast.finish() + broadcast.close() def test_publish_media_cut_and_seek(): @@ -148,7 +148,7 @@ def test_publish_media_cut_and_seek(): media.seek(42) media.finish() - broadcast.finish() + broadcast.close() def test_video_properties_use_defaulted_fields(): @@ -157,7 +157,7 @@ def test_video_properties_use_defaulted_fields(): assert properties.display is None assert properties.flip is None broadcast.set_video_properties(properties) - broadcast.finish() + broadcast.close() def test_audio_rejects_bad_init_bytes(): @@ -303,12 +303,19 @@ async def test_catalog_update_on_new_track(): break -def test_finish_closes_producer(): +def test_close_twice_is_a_noop(): broadcast = moq.BroadcastProducer() _media = broadcast.publish_audio(moq.AudioFormat.OPUS, opus_head()) - broadcast.finish() + broadcast.close() + broadcast.close() with pytest.raises(Exception): + broadcast.publish_audio(moq.AudioFormat.OPUS, opus_head()) + + +def test_finish_is_deprecated(): + broadcast = moq.BroadcastProducer() + with pytest.deprecated_call(): broadcast.finish() @@ -330,7 +337,7 @@ def test_publish_lifecycle(): track = broadcast.publish_track("status") track.write_frame(b'{"cmd": "ready"}', 0) track.finish() - broadcast.finish() + broadcast.close() async def test_publish_track_info_and_subscription(): @@ -520,7 +527,7 @@ async def test_dynamic_broadcast_request(): assert frame.payload == payload assert frame.timestamp_us == 20_000 track.finish() - served.finish() + served.close() async def test_dynamic_broadcast_request_can_reject(): @@ -1075,7 +1082,7 @@ def test_encode_audio_with_opus_object(): producer.write(moq.AudioFrame(timestamp_us=0, data=bytes(960 * 4))) assert producer.name == "mic" producer.finish() - broadcast.finish() + broadcast.close() async def test_decode_video_format(): @@ -1129,7 +1136,7 @@ async def test_decode_video_format(): i420.cancel() packed.cancel() video.finish() - broadcast.finish() + broadcast.close() async def test_broadcast_is_reachable_only_while_announced(): @@ -1159,7 +1166,7 @@ async def test_broadcast_is_reachable_only_while_announced(): await asyncio.wait_for(consumer.request_broadcast("live"), timeout=5.0) announced.cancel() track.finish() - broadcast.finish() + broadcast.close() async def test_announced_pattern_captures(): @@ -1180,8 +1187,8 @@ async def test_announced_pattern_captures(): announced.cancel() dynamic.cancel() - audio.finish() - chat.finish() + audio.close() + chat.close() async def test_dynamic_serves_a_request_under_a_prefix(): @@ -1196,7 +1203,7 @@ async def test_dynamic_serves_a_request_under_a_prefix(): request.accept(served) await asyncio.wait_for(pending, timeout=5.0) dynamic.cancel() - served.finish() + served.close() async def test_dynamic_and_json_handles_are_async_context_managers(): @@ -1235,4 +1242,4 @@ async def assert_cancelled(awaitable) -> None: snapshot.finish() stream.finish() track.finish() - broadcast.finish() + broadcast.close() diff --git a/py/moq-rs/tests/test_server.py b/py/moq-rs/tests/test_server.py index 170129306b..3dd07bc8a9 100644 --- a/py/moq-rs/tests/test_server.py +++ b/py/moq-rs/tests/test_server.py @@ -72,7 +72,7 @@ async def accept_loop() -> None: except asyncio.CancelledError: pass media.finish() - broadcast.finish() + broadcast.close() async def test_client_reconnects_and_resumes_announcements(): @@ -127,7 +127,7 @@ async def accept_loop() -> None: async for announcement in client.announced(): assert announcement.prefix == "after-reconnect" break - broadcast.finish() + broadcast.close() finally: accept_task.cancel() try: @@ -287,7 +287,7 @@ async def test_serve_helper_accepts_clients(): await serve_task except asyncio.CancelledError: pass - broadcast.finish() + broadcast.close() async def test_broadcast_route_over_wire(): @@ -317,7 +317,7 @@ async def test_broadcast_route_over_wire(): await serve_task except asyncio.CancelledError: pass - broadcast.finish() + broadcast.close() async def test_route_update_observes_restart(): diff --git a/quest/m1/README.md b/quest/m1/README.md index 93cb6d3458..530c0fddaa 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -31,6 +31,7 @@ transport, benchmark tooling); worktrees isolate commits, not semantics. - [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 - [Broadcast close](/quest/m1/broadcast-close/README.md) - `close()` is the one way to end a broadcast in every language, a permanent retraction that leaves in-flight tracks alone +- [Remove finish](/quest/m1/broadcast-remove.md) - on dev, the deprecated broadcast end APIs are gone and `closed()` carries no cause - [Relay peer set](/quest/m1/relay-peer-set.md) - a wire consumer tells a client hop from a peer hop, and every mesh credential can mark a peer - [CLI import clock](/quest/m1/cli-import-clock.md) - fMP4, TS, and FLV imports publish on the shared broadcast clock across restarts - [Native clock fixtures](/quest/m1/native-clock-fixtures.md) - CI drives native capture through clock edge cases and asserts the published timestamps diff --git a/quest/m1/broadcast-close/README.md b/quest/m1/broadcast-close/README.md index 22d5bf99ba..55721f39a9 100644 --- a/quest/m1/broadcast-close/README.md +++ b/quest/m1/broadcast-close/README.md @@ -1,4 +1,4 @@ -# Broadcast close +# [S] Broadcast close ## Goal @@ -34,14 +34,8 @@ object ended", not "the path went offline": announcements say whether a path is live, and a path can be announced again by a new object. Caches and moq-hls use it to tell a live object from a replaced one at the same path. -Stage it as the children below: Rust first, then the bindings, then the `dev` -removal. - -## Quests - -- [Rust close](/quest/m1/broadcast-close/rust.md) - moq-net gains `close()`, deprecates `finish`/`abort`/`is_finished`, and every Rust and JS caller moves over -- [Binding close](/quest/m1/broadcast-close/bindings.md) - moq-ffi, libmoq, and every wrapper expose `close()` and deprecate `finish` -- [Remove finish](/quest/m1/broadcast-close/remove.md) - on dev, the deprecated broadcast end APIs are gone and `closed()` carries no cause +The `dev` removal is [Remove finish](/quest/m1/broadcast-remove.md), which +waits for this line to reach `main`. ## Related diff --git a/quest/m1/broadcast-close/bindings.md b/quest/m1/broadcast-close/bindings.md deleted file mode 100644 index 86437ec9e4..0000000000 --- a/quest/m1/broadcast-close/bindings.md +++ /dev/null @@ -1,29 +0,0 @@ -# [M] Binding close - -## Goal - -Every binding ends a broadcast with `close()`, mirroring Rust, and its -`finish` is deprecated. - -## Plan - -- moq-ffi: add `MoqBroadcastProducer::close()` next to `finish()` in - `rs/moq-ffi/src/producer.rs`. It closes the broadcast, then the catalog, as - `finish` does, so a catalog error cannot stop the broadcast from ending. - Deprecate `finish`. -- libmoq: export `moq_publish_close` and deprecate `moq_publish_finish`. Move - `cpp/obs/src/moq-output.cpp` and the C tests over. -- Wrappers: `py/moq-rs` `BroadcastProducer.close()` (its `finish` docstring - wrongly says it closes the tracks), swift `Broadcast.close()`, and the Go - `Publish.Close()`. Kotlin and Dart only alias the generated type, so they - pick `close` up from moq-ffi. -- Move the binding tests over, keeping one test per binding that a second - `close` errors or no-ops, whichever Rust settles on. -- Update `doc/lib/{py,swift,kt,go,dart,c}`, including `doc/lib/go/index.md`'s - `broadcast.Finish()` sample. -- Fix the moq-ffi `origin.rs` doc comment that points users at a - `broadcast.closed()` the bindings don't have. - -## Required - -- [Rust close](/quest/m1/broadcast-close/rust.md) - the binding forwards to `broadcast::Producer::close` diff --git a/quest/m1/broadcast-close/rust.md b/quest/m1/broadcast-close/rust.md deleted file mode 100644 index b2f8ea9acc..0000000000 --- a/quest/m1/broadcast-close/rust.md +++ /dev/null @@ -1,49 +0,0 @@ -# [M] Rust close - -## Goal - -`moq_net::broadcast::Producer::close()` is the one way to end a broadcast, and -no Rust or JS code in the repository calls a deprecated end API. - -## Plan - -In `rs/moq-net/src/model/broadcast.rs`: - -- Add `close(&self)`: set `closing`, close the liveness token, and retire the - announcer, as `finish` does today without setting `finished`. Borrow, like - `finish`: any clone ends it. -- Once a broadcast has ended, by `close()` or its last producer dropping, every - new lookup on any consumer answers `Unroutable`: `Consumer::track`, and the - requests still pending for names nothing served. That is what a fresh - `request_broadcast` for the path answers, so a consumer can't tell a raw - handle from one reached through an origin, or a local source from a remote - one. Today the same lookup answers `NotFound` while clones live and `Dropped` - after. Tracks already read are untouched and end on their own FIN or reset. - This changes the error a published API returns; say so in the PR. -- Document `broadcast::Producer::consume()` as a view of this one publisher: a - new publisher at the path is never spliced into it, so a consumer that should - not care about its source goes through an origin. -- Mark `finish`, `abort`, and `Consumer::is_finished` `#[deprecated]` and - `#[doc(hidden)]` per `rs/CLAUDE.md`. `finish` forwards to `close`. Leave - `abort`'s behavior alone until the `dev` removal. -- Drop the "dropped without finish()" warning in `Drop for Alive`. -- `SourceGuard` (the lite and IETF subscribers' source handle) closes on both - a graceful end and a drop, since the abort it records on drop reaches no one - past this hop. Check the IETF `Detach::Abrupt` path still ends the source. - -Callers to move (production): `model/origin.rs` (front end), -`lite/subscriber.rs` (`Route::finish`), `ietf/subscriber.rs`, moq-srt -`ts.rs`, moq-rtmp `server.rs` and `dial.rs`, moq-stats `produce.rs`, moq-gst -`sink/imp.rs`, moq-rtc `server/mod.rs`, moq-transcode `lib.rs`, moq-relay -`cluster.rs` (gossip registration), moq-boy `main.rs`, and moq-tokio's -`clock` and `chat` examples. Tests and benches across moq-net, moq-tokio, -moq-hls, moq-bench, and moq-gst move too. - -moq-srt and moq-rtmp `Publisher::abort` drop the broadcast instead of ending -it; closing it explicitly makes that path deterministic. - -In `js/net/src/broadcast.ts`, `Producer.close()` and `Consumer.close()` stop -taking `abort`: deprecate the parameter in the JSDoc and move the two origin -teardown callers in `origin.ts`. JS `close()` already blocks re-announcing. - -Update `doc/lib/rs/moq-net.md`, which says the route retracts on `finish()`. diff --git a/quest/m1/broadcast-close/remove.md b/quest/m1/broadcast-remove.md similarity index 56% rename from quest/m1/broadcast-close/remove.md rename to quest/m1/broadcast-remove.md index 805896a6ee..c48da3351c 100644 --- a/quest/m1/broadcast-close/remove.md +++ b/quest/m1/broadcast-remove.md @@ -15,7 +15,11 @@ This is a published API break, so it targets `dev`. `()` rather than an `Error` cause. - Remove the deprecated binding `finish` methods, `moq_publish_finish`, and JS's `close(abort)` parameter. +- Kotlin has no generated `close()`: it would collide with `AutoCloseable.close()`, + so `rs/moq-ffi/uniffi.toml` excludes it. Kotlin's `close()` releases the handle, + which ends the broadcast only once no `dynamic()` handle remains. Removing + `finish` leaves Kotlin without a forced end; decide whether it needs one. ## Required -- [Binding close](/quest/m1/broadcast-close/bindings.md) - every binding already has `close()` to move to +- [Broadcast close](/quest/m1/broadcast-close/README.md) - the deprecations this removes, which must reach `main` and then `dev` first diff --git a/rs/libmoq/README.md b/rs/libmoq/README.md index 635024d478..2ea1d55bb9 100644 --- a/rs/libmoq/README.md +++ b/rs/libmoq/README.md @@ -63,7 +63,7 @@ int32_t moq_origin_announced_cancel(uint32_t announced); // Publishing int32_t moq_publish_announce(uint32_t broadcast, const moq_route *route); int32_t moq_publish_unannounce(uint32_t broadcast); -int32_t moq_publish_finish(uint32_t broadcast); +int32_t moq_publish_close(uint32_t broadcast); int32_t moq_publish_audio(uint32_t broadcast, const moq_audio_init *config); int32_t moq_publish_video(uint32_t broadcast, const moq_video_init *config); int32_t moq_publish_container(uint32_t broadcast, const moq_container_init *config); diff --git a/rs/libmoq/c-tests/decode-output.c b/rs/libmoq/c-tests/decode-output.c index 585f107bf2..057f33e62d 100644 --- a/rs/libmoq/c-tests/decode-output.c +++ b/rs/libmoq/c-tests/decode-output.c @@ -284,8 +284,8 @@ int main(void) { fail("error: moq_consume_close failed (%s)\n", moq_error()); if (moq_encode_video_finish((uint32_t)producer) < 0) fail("error: moq_encode_video_finish failed (%s)\n", moq_error()); - if (moq_publish_finish((uint32_t)broadcast) < 0) - fail("error: moq_publish_finish failed (%s)\n", moq_error()); + if (moq_publish_close((uint32_t)broadcast) < 0) + fail("error: moq_publish_close failed (%s)\n", moq_error()); if (moq_origin_close((uint32_t)origin) < 0) fail("error: moq_origin_close failed (%s)\n", moq_error()); diff --git a/rs/libmoq/src/api.rs b/rs/libmoq/src/api.rs index 66e353a052..409c80ed35 100644 --- a/rs/libmoq/src/api.rs +++ b/rs/libmoq/src/api.rs @@ -1543,7 +1543,7 @@ pub extern "C" fn moq_origin_create() -> i32 { /// /// The broadcast is invisible and unroutable, on this origin and its peers /// alike, until [moq_publish_announce]. Fill it with the `moq_publish_*` -/// functions, then announce it. [moq_publish_finish] unpublishes immediately. +/// functions, then announce it. [moq_publish_close] ends it for good. /// /// Returns a non-zero broadcast handle on success, or a negative code on failure. /// @@ -1923,20 +1923,29 @@ pub extern "C" fn moq_publish_unannounce(broadcast: u32) -> i32 { }) } -/// Finish a broadcast and release it, ending its catalog cleanly. +/// End a broadcast for good and release its handle. /// -/// Subscribers see a normal end of stream rather than an error, and the origin unpublishes -/// the path immediately. +/// The origin retracts the path immediately and serves no new tracks; tracks already +/// subscribed carry on to their own end. The handle is invalid afterwards, so closing +/// it again fails like any unknown handle. /// /// Returns a zero on success, or a negative code on failure. #[unsafe(no_mangle)] -pub extern "C" fn moq_publish_finish(broadcast: u32) -> i32 { +pub extern "C" fn moq_publish_close(broadcast: u32) -> i32 { ffi::enter(move || { let broadcast = ffi::parse_id(broadcast)?; - State::lock().publish.finish(broadcast) + State::lock().publish.close(broadcast) }) } +/// 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/publish.rs b/rs/libmoq/src/publish.rs index fd7db2f681..7b93c083e0 100644 --- a/rs/libmoq/src/publish.rs +++ b/rs/libmoq/src/publish.rs @@ -162,9 +162,8 @@ impl Publish { Ok((&mut broadcast.producer, &mut broadcast.catalog)) } - /// Cleanly finish the broadcast and finalize the catalog stream, so subscribers - /// see a normal end rather than [`moq_net::Error::Dropped`]. - pub fn finish(&mut self, broadcast: Id) -> Result<(), Error> { + /// End the broadcast for good and release it, finalizing the catalog stream. + pub fn close(&mut self, broadcast: Id) -> Result<(), Error> { let Broadcast { producer, mut catalog, @@ -183,9 +182,8 @@ impl Publish { } guard.commit()?; } - // Finish the broadcast first so the clean end reaches subscribers even if - // finalizing the catalog fails. - producer.finish(); + // Close the broadcast first so it ends even if finalizing the catalog fails. + producer.close(); catalog.finish()?; Ok(()) } diff --git a/rs/libmoq/src/test.rs b/rs/libmoq/src/test.rs index 535d3574d8..d48de75362 100644 --- a/rs/libmoq/src/test.rs +++ b/rs/libmoq/src/test.rs @@ -367,7 +367,7 @@ fn publish_media_lifecycle() { let origin = id(moq_origin_create()); let broadcast = publish_broadcast(origin, b"publish-media-lifecycle"); let _guard = Guard(Some(|| { - moq_publish_finish(broadcast); + moq_publish_close(broadcast); })); let init = opus_head(); @@ -379,7 +379,7 @@ fn publish_media_lifecycle() { assert_eq!(ret, 0, "moq_publish_media_frame should succeed"); assert_eq!(moq_publish_media_finish(media), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); } #[test] @@ -389,7 +389,7 @@ fn publish_media_rejects_a_null_config() { assert!(unsafe { moq_publish_audio(broadcast, std::ptr::null()) } < 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -424,7 +424,7 @@ fn container_and_media_handles_are_not_interchangeable() { assert_eq!(moq_publish_container_finish(container), 0); assert_eq!(moq_publish_media_finish(media), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -511,7 +511,7 @@ fn publish_media_labels_config_without_naming_track() { assert_eq!(moq_consume_close(consume), 0); assert_eq!(moq_publish_media_finish(media1), 0); assert_eq!(moq_publish_media_finish(media2), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -621,7 +621,7 @@ fn publish_media_owns_its_rendition_before_the_first_keyframe() { "finishing the media track releases its rendition name" ); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -667,7 +667,7 @@ fn publish_video_config_replaces_its_own_rendition() { "the name is free once the caller removes its rendition" ); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -690,7 +690,7 @@ fn publish_catalog_config_null_pointer() { -6, "null config should return InvalidPointer (-6)" ); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); } #[test] @@ -891,7 +891,7 @@ fn publish_catalog_roundtrip() { assert_eq!(moq_consume_catalog_cancel(catalog_task), 0); assert_eq!(catalog_cb.recv_terminal(), 0, "catalog close delivers terminal 0"); assert_eq!(moq_consume_close(consume), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -963,13 +963,13 @@ fn a_half_specified_coded_size_round_trips() { assert_eq!(moq_consume_catalog_cancel(forwarded_task), 0); assert_eq!(forwarded_cb.recv_catalog_terminal(), 0); assert_eq!(moq_consume_close(forwarded), 0); - assert_eq!(moq_publish_finish(forward), 0); + assert_eq!(moq_publish_close(forward), 0); assert_eq!(moq_consume_catalog_free(catalog), 0); assert_eq!(moq_consume_catalog_cancel(catalog_task), 0); assert_eq!(catalog_cb.recv_catalog_terminal(), 0); assert_eq!(moq_consume_close(consume), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -1065,7 +1065,7 @@ fn raw_loc_video_uses_the_declared_catalog_container() { assert_eq!(catalog_cb.recv_terminal(), 0); assert_eq!(moq_consume_close(consume), 0); assert_eq!(moq_publish_track_finish(track), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -1129,7 +1129,7 @@ fn cmaf_catalog_container_carries_its_init_segment() { assert_eq!(moq_consume_catalog_cancel(catalog_task), 0); assert_eq!(catalog_cb.recv_terminal(), 0); assert_eq!(moq_consume_close(consume), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -1189,7 +1189,7 @@ fn unpublishable_catalog_containers_are_rejected() { "cmaf without an init segment should return InvalidPointer (-6)" ); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -1344,7 +1344,7 @@ fn catalog_section_roundtrip() { assert_eq!(moq_consume_catalog_cancel(catalog_task), 0); assert_eq!(catalog_cb.recv_terminal(), 0, "catalog close delivers terminal 0"); assert_eq!(moq_consume_close(consume), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -1391,7 +1391,7 @@ fn publish_track_with_info_rejects_invalid_timescale() { }; assert!(unsafe { moq_publish_track(broadcast, name.as_ptr() as *const c_char, name.len(), &info) } < 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); } #[test] @@ -1491,7 +1491,7 @@ fn raw_track_publish_consume() { assert_eq!(moq_publish_track_finish(track), 0); assert!(moq_publish_track_finish(track) < 0, "double-close should fail"); assert_eq!(moq_consume_close(consume), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -1553,7 +1553,7 @@ fn raw_track_datagram_publish_consume() { assert!(moq_consume_datagrams_cancel(consumer) < 0, "double-close should fail"); assert_eq!(moq_publish_track_finish(track), 0); assert_eq!(moq_consume_close(consume), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -1572,7 +1572,7 @@ fn raw_track_sparse_groups_and_known_end() { assert_eq!(moq_publish_group_finish(group), 0); assert!(moq_publish_track_group_at(track, 5) < 0); assert_eq!(moq_publish_track_finish(track), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); } #[test] @@ -1587,7 +1587,7 @@ fn raw_track_and_group_abort_consume_their_handles() { assert!(moq_publish_group_finish(group) < 0); assert_eq!(moq_publish_track_abort(track, 410), 0); assert!(moq_publish_track_finish(track) < 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); } #[test] @@ -1673,7 +1673,7 @@ fn raw_track_subscription_options_and_update() { assert_eq!(frame_cb.recv_terminal(), 0); assert_eq!(moq_publish_track_finish(track), 0); assert_eq!(moq_consume_close(consume), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -1739,7 +1739,7 @@ fn json_snapshot_publish_consume() { "double-close should fail" ); assert_eq!(moq_consume_close(consume), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -1799,7 +1799,7 @@ fn json_stream_publish_consume() { assert_eq!(moq_publish_json_stream_finish(producer), 0); assert!(moq_publish_json_stream_finish(producer) < 0, "double-close should fail"); assert_eq!(moq_consume_close(consume), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -1807,13 +1807,26 @@ fn json_stream_publish_consume() { fn close_invalid_or_zero_ids() { assert!(moq_origin_close(9999) < 0); assert!(moq_session_close(9999) < 0); - assert!(moq_publish_finish(9999) < 0); + assert!(moq_publish_close(9999) < 0); assert!(moq_consume_close(9999) < 0); assert!(moq_consume_frame_free(9999) < 0); assert!(moq_origin_close(0) < 0); assert!(moq_session_close(0) < 0); - assert!(moq_publish_finish(0) < 0); + assert!(moq_publish_close(0) < 0); +} + +#[test] +fn publish_close_releases_the_handle() { + let origin = id(moq_origin_create()); + 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); } #[test] @@ -1865,7 +1878,7 @@ fn announced_free_lifecycle() { ann_cb.recv_terminal(); assert_eq!(moq_origin_close(origin), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); } #[test] @@ -1882,7 +1895,7 @@ fn double_close_all_resource_types() { assert_eq!(moq_publish_media_finish(media), 0); assert!(moq_publish_media_finish(media) < 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); let origin = id(moq_origin_create()); let path = b"double-close-test"; @@ -1922,7 +1935,7 @@ fn double_close_all_resource_types() { assert_eq!(moq_consume_close(consume), 0); assert_eq!(moq_publish_media_finish(media), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -1961,7 +1974,7 @@ fn media_cut_bounds_audio_groups() { assert!(moq_publish_media_seek(9999, 0) < 0); assert_eq!(moq_publish_media_finish(media), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -1970,7 +1983,7 @@ fn unknown_format() { let origin = id(moq_origin_create()); let broadcast = publish_broadcast(origin, b"unknown-format"); let _guard = Guard(Some(|| { - moq_publish_finish(broadcast); + moq_publish_close(broadcast); })); // A format is an enum now, so the only bad value C can still supply is an out-of-range @@ -2025,7 +2038,7 @@ fn local_announce() { assert_eq!(moq_origin_announced_cancel(announced_task), 0); assert_eq!(cb.recv_terminal(), 0, "announced close delivers terminal 0"); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -2072,8 +2085,8 @@ fn announced_filters_patterns_and_reports_captures() { assert_eq!(moq_origin_announced_free(announced_id), 0); assert_eq!(moq_origin_announced_cancel(announced_task), 0); assert_eq!(cb.recv_terminal(), 0); - assert_eq!(moq_publish_finish(audio), 0); - assert_eq!(moq_publish_finish(chat), 0); + assert_eq!(moq_publish_close(audio), 0); + assert_eq!(moq_publish_close(chat), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -2125,7 +2138,7 @@ fn announced_deactivation() { assert_eq!(moq_origin_announced_cancel(announced_task), 0); assert_eq!(cb.recv_terminal(), 0, "announced close delivers terminal 0"); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -2166,7 +2179,7 @@ fn create_broadcast_is_unroutable_until_announced() { assert_eq!(moq_origin_announced_cancel(announced_task), 0); assert_eq!(cb.recv_terminal(), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -2184,7 +2197,7 @@ fn announce_accepts_an_anonymous_hop() { has_cold: false, }; assert_eq!(unsafe { moq_publish_announce(broadcast, &route) }, 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -2233,7 +2246,7 @@ fn dynamic_serves_a_request_under_a_prefix() { assert_eq!(moq_origin_dynamic_cancel(dynamic), 0); assert!(moq_origin_dynamic_cancel(dynamic) < 0, "double-cancel should fail"); assert_eq!(cb.recv_terminal(), 0); - assert_eq!(moq_publish_finish(served), 0); + assert_eq!(moq_publish_close(served), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -2363,7 +2376,7 @@ fn track_demand_follows_subscribers() { ); assert_eq!(moq_consume_close(consume), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -2396,7 +2409,7 @@ fn track_demand_reports_current_state_before_close() { ); assert_eq!(moq_publish_track_finish(track), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -2419,7 +2432,7 @@ fn track_demand_reports_an_abort() { assert_eq!(protocol.kind, moq_protocol_kind::MOQ_PROTOCOL_KIND_APP as u32); assert_eq!(protocol.code, 64 + 7); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -2456,7 +2469,7 @@ fn media_demand_refuses_a_container() { assert_eq!(cb.recv_terminal(), 0); assert_eq!(moq_publish_container_finish(container), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -2559,7 +2572,7 @@ fn dynamic_serves_track_requests() { assert_eq!(moq_publish_track_finish(track), 0); assert_eq!(demand_cb.recv_terminal(), 0); assert_eq!(moq_consume_close(consume), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -2603,7 +2616,7 @@ fn dynamic_track_request_publishes_media() { assert_eq!(demand_cb.recv(), moq_demand::MOQ_DEMAND_UNUSED as i32); assert_eq!(moq_publish_media_finish(media), 0); assert_eq!(demand_cb.recv_terminal(), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(request_cb.recv_terminal(), 0); assert_eq!(moq_consume_close(consume), 0); assert_eq!(moq_origin_close(origin), 0); @@ -2688,7 +2701,7 @@ fn track_dynamic_serves_a_fetch_miss() { assert_eq!(moq_publish_dynamic_cancel(dynamic), 0); assert_eq!(group_cb.recv_terminal(), 0); assert_eq!(moq_publish_track_finish(track), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -2723,7 +2736,7 @@ fn track_dynamic_serves_a_fetch_from_frame_start() { assert_eq!(moq_publish_dynamic_cancel(dynamic), 0); assert_eq!(group_cb.recv_terminal(), 0); assert_eq!(moq_publish_track_finish(track), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -2757,7 +2770,7 @@ fn track_request_dynamic_survives_accept() { assert_eq!(moq_publish_dynamic_cancel(track_dynamic), 0); assert_eq!(group_cb.recv_terminal(), 0); assert_eq!(moq_publish_track_finish(track), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(request_cb.recv_terminal(), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -2855,7 +2868,7 @@ fn local_publish_consume() { assert_eq!(catalog_cb.recv_terminal(), 0, "catalog close delivers terminal 0"); assert_eq!(moq_consume_close(consume), 0); assert_eq!(moq_publish_media_finish(media), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -2912,7 +2925,7 @@ fn consume_announced_local() { assert_eq!(catalog_cb.recv_terminal(), 0, "catalog close delivers terminal 0"); assert_eq!(moq_consume_close(consume), 0); assert_eq!(moq_publish_media_finish(media), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -3014,8 +3027,8 @@ fn consume_audio_follows_a_sibling_broadcast_reference() { assert_eq!(catalog_cb.recv_terminal(), 0, "catalog close delivers terminal 0"); assert_eq!(moq_consume_close(consume), 0); assert_eq!(moq_publish_media_finish(media), 0); - assert_eq!(moq_publish_finish(broadcast), 0); - assert_eq!(moq_publish_finish(source), 0); + assert_eq!(moq_publish_close(broadcast), 0); + assert_eq!(moq_publish_close(source), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -3142,7 +3155,7 @@ fn video_publish_consume() { assert_eq!(catalog_cb.recv_terminal(), 0, "catalog close delivers terminal 0"); assert_eq!(moq_consume_close(consume), 0); assert_eq!(moq_publish_media_finish(media), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -3197,7 +3210,7 @@ fn audio_raw_publish() { "a finished producer should take no more frames" ); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -3252,7 +3265,7 @@ fn audio_raw_publish_frame_durations() { assert_eq!(moq_encode_audio_finish(id(encode(b"default", 0))), 0, "0 = 20 ms"); assert!(encode(b"rounded", 2_000) < 0, "2 ms is not an opus frame duration"); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -3355,7 +3368,7 @@ fn video_raw_publish_consume() { assert_eq!(catalog_cb.recv_catalog_terminal(), 0); assert_eq!(moq_consume_close(consume), 0); assert_eq!(moq_encode_video_finish(producer), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -3493,7 +3506,7 @@ fn decode_first_frame(output: &moq_video_decoder_output) -> (u32, u32, usize) { assert_eq!(catalog_cb.recv_catalog_terminal(), 0); assert_eq!(moq_consume_close(consume), 0); assert_eq!(moq_encode_video_finish(producer), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); result } @@ -3578,7 +3591,7 @@ fn video_raw_publish_from_many_threads() { .join() .unwrap(); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -3674,7 +3687,7 @@ fn a_stalled_encode_does_not_block_unrelated_calls() { assert_eq!(moq_origin_close(id(created)), 0); assert_eq!(moq_encode_video_finish(stalled), 0); assert_eq!(moq_encode_video_finish(other), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -3712,7 +3725,7 @@ fn video_raw_publish_rejects_frame_size_mismatch() { assert!(unsafe { moq_encode_video_frame(producer, &frame) } < 0); assert_eq!(moq_encode_video_finish(producer), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -3801,7 +3814,7 @@ fn video_raw_publish_rejects_invalid_config() { assert!(moq_encode_video_bitrate(0, 1_000_000) < 0); assert!(moq_encode_video_finish(0) < 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -3901,7 +3914,7 @@ fn video_raw_decode() { } assert_eq!(moq_consume_close(consume), 0); assert_eq!(moq_publish_media_finish(media), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -3961,7 +3974,7 @@ fn multiple_frames_ordering() { ); assert_eq!(moq_consume_close(consume), 0); assert_eq!(moq_publish_media_finish(media), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -4010,7 +4023,7 @@ fn catalog_update_on_new_track() { assert_eq!(moq_consume_close(consume), 0); assert_eq!(moq_publish_media_finish(media1), 0); assert_eq!(moq_publish_media_finish(media2), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -4549,7 +4562,7 @@ fn bandwidth_reservations_split_the_estimate() { let _ = first_cb.recv_terminal(); let _ = second_cb.recv_terminal(); assert_eq!(moq_consume_close(consume), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -4592,7 +4605,7 @@ fn bandwidth_handles_share_the_registry() { let _ = first_cb.recv_terminal(); let _ = second_cb.recv_terminal(); assert_eq!(moq_consume_close(consume), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -4642,7 +4655,7 @@ fn encode_video_bitrate_caps_the_reservation() { assert_eq!(moq_encode_video_finish(producer), 0); assert_eq!(moq_bandwidth_close(bandwidth), 0); assert_eq!(moq_consume_close(consume), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -4761,7 +4774,7 @@ fn server_accepts_a_session() { "the server handle is gone after its terminal callback" ); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(served), 0); assert_eq!(moq_origin_close(received), 0); } @@ -4931,7 +4944,7 @@ fn json_tracks_are_advertised_in_the_catalog() { assert_eq!(moq_publish_json_stream_finish(stream), 0); assert!(published_catalog(broadcast).json.tracks.is_empty()); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -4979,7 +4992,7 @@ fn binary_snapshot_is_advertised_and_delivered() { assert!(published_catalog(broadcast).binary.tracks.is_empty()); assert_eq!(moq_consume_close(consume), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -5025,7 +5038,7 @@ fn binary_stream_is_advertised_and_delivered() { assert!(published_catalog(broadcast).binary.tracks.is_empty()); assert_eq!(moq_consume_close(consume), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } @@ -5087,6 +5100,6 @@ fn data_track_names_cannot_collide() { ); assert_eq!(moq_publish_json_snapshot_finish(first), 0); - assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); } diff --git a/rs/moq-bench/src/connection.rs b/rs/moq-bench/src/connection.rs index 1502ab5c87..383f7be991 100644 --- a/rs/moq-bench/src/connection.rs +++ b/rs/moq-bench/src/connection.rs @@ -799,7 +799,7 @@ mod tests { task.await.unwrap().unwrap(); assert_eq!(stats.groups_recv.load(Ordering::Relaxed), 1); - broadcast.finish(); + broadcast.close(); } /// The relay fails a group it gave up on (`Error::Lagged` once a subscriber @@ -850,7 +850,7 @@ mod tests { write_group(&mut track); wait_for(&stats.groups_recv, 3).await; track.finish().unwrap(); - broadcast.finish(); + broadcast.close(); task.await .unwrap() diff --git a/rs/moq-boy/src/main.rs b/rs/moq-boy/src/main.rs index a167fbec44..df54bf2baa 100644 --- a/rs/moq-boy/src/main.rs +++ b/rs/moq-boy/src/main.rs @@ -305,9 +305,8 @@ async fn run(config: &Config) -> Result<()> { res = input::handle_viewers(&viewer_consumer, &cmd_tx) => res, }; - // Cleanly close the broadcast so subscribers see a normal end rather than - // Error::Dropped. - broadcast.finish(); + // Close the broadcast now, even if the emulator thread still holds a clone. + broadcast.close(); result } diff --git a/rs/moq-ffi/src/origin.rs b/rs/moq-ffi/src/origin.rs index 1f325f1692..253d5a749e 100644 --- a/rs/moq-ffi/src/origin.rs +++ b/rs/moq-ffi/src/origin.rs @@ -504,7 +504,7 @@ impl MoqAnnounceUpdate { impl MoqAnnouncedBroadcast { /// Wait until the broadcast is announced. Returns `Closed` if cancelled or the origin is closed. /// - /// Use `broadcast.closed()` to learn when the broadcast ends. + /// Its end arrives as an inactive [`MoqAnnounceUpdate`] on the origin's announcements. pub async fn available(&self) -> Result, MoqError> { self.task.run(|mut state| async move { state.available().await }).await } diff --git a/rs/moq-ffi/src/producer.rs b/rs/moq-ffi/src/producer.rs index 04822ca67c..c1117c2eeb 100644 --- a/rs/moq-ffi/src/producer.rs +++ b/rs/moq-ffi/src/producer.rs @@ -170,7 +170,7 @@ impl MoqBroadcastProducer { } /// Run `f` against the open broadcast and catalog. Errors with - /// [`MoqError::Closed`] if `finish()` has already run. Used by + /// [`MoqError::Closed`] if `close()` has already run. Used by /// sibling modules (e.g. `audio`) that need joint access. pub(crate) fn with_state( &self, @@ -470,18 +470,27 @@ impl MoqBroadcastProducer { })) } - /// Finish this publisher, finalizing the catalog stream and cleanly closing the - /// broadcast so subscribers see a normal end rather than `Error::Dropped`. - pub fn finish(&self) -> Result<(), MoqError> { + /// End the broadcast for good: retract it, serve no new tracks, and finalize the catalog. + /// + /// Tracks already subscribed carry on to their own end. Every later call on this + /// producer fails with `Closed`; closing again is a no-op. + pub fn close(&self) -> Result<(), MoqError> { let _guard = crate::ffi::enter(); + // Hold the lock through shutdown so a concurrent close() returns only once it is done. let mut guard = self.state.lock().unwrap(); - let mut state = guard.take().ok_or(MoqError::Closed)?; - // Finish the broadcast first so the clean end reaches subscribers even if - // finalizing the catalog fails. - state.broadcast.finish(); + let Some(mut state) = guard.take() else { + return Ok(()); + }; + // Close the broadcast first so it ends even if finalizing the catalog fails. + state.broadcast.close(); 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 265dee6162..ba0e2dc0bb 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -332,7 +332,7 @@ async fn announced_route_keeps_cold_cost_on_reannounce() { assert_eq!(back.cost, moq_net::origin::Cost { warm: 0, cold: 9 }); assert_eq!(MoqRoute::from(back), route); - broadcast.finish().unwrap(); + broadcast.close().unwrap(); } #[test] @@ -347,7 +347,7 @@ fn publish_media_lifecycle() { }) .unwrap(); media.finish().unwrap(); - broadcast.finish().unwrap(); + broadcast.close().unwrap(); } #[tokio::test] @@ -467,7 +467,7 @@ async fn raw_audio_activity() { assert_eq!(resumed.timestamp_us, RESUMED_TIMESTAMP_US); audio.finish().unwrap(); - broadcast.finish().unwrap(); + broadcast.close().unwrap(); } /// `frame_duration_us` is microseconds so Opus' 2.5 ms frame survives the trip, where @@ -509,7 +509,7 @@ async fn raw_audio_frame_durations() { "2 ms is not an opus frame duration" ); - broadcast.finish().unwrap(); + broadcast.close().unwrap(); } #[tokio::test] @@ -1482,7 +1482,7 @@ async fn create_broadcast_is_invisible_until_announced() { .expect("the cursor is still open"); assert_eq!(update.prefix(), "live"); assert!(update.active()); - broadcast.finish().unwrap(); + broadcast.close().unwrap(); } /// Waiting for an exact path must hand the broadcast back named by that path, the base a @@ -1509,7 +1509,7 @@ async fn announced_broadcast_keeps_the_requested_path() { .unwrap(); assert_eq!(requested.inner().info().path.as_str(), "a/pub"); - broadcast.finish().unwrap(); + broadcast.close().unwrap(); } /// A catalog rendition may name a sibling broadcast (`./source`), and the track then lives @@ -1550,8 +1550,8 @@ async fn decode_audio_follows_a_sibling_broadcast_reference() { "the catalog broadcast does not serve the track itself" ); - catalog.finish().unwrap(); - source.finish().unwrap(); + catalog.close().unwrap(); + source.close().unwrap(); } /// Announcement filters do not re-root the origin, so relative broadcast references @@ -1605,8 +1605,8 @@ async fn announced_broadcasts_resolve_siblings_under_the_prefix() { .expect("timed out subscribing on the resolved broadcast") .unwrap(); - catalog.finish().unwrap(); - source.finish().unwrap(); + catalog.close().unwrap(); + source.close().unwrap(); } #[tokio::test] @@ -1642,7 +1642,7 @@ async fn announced_filters_patterns_and_reports_captures() { assert_eq!(update.captures(), Some(vec!["alice".into()])); assert!(update.active()); - chat.finish().unwrap(); + chat.close().unwrap(); } /// A `.`-named broadcast is listed only when the config opts in or the prefix names it. @@ -1731,8 +1731,8 @@ async fn resolve_returns_a_broadcast_that_resolves_further_references() { .unwrap(); assert_eq!(back.inner().info().path.as_str(), "a/pub"); - catalog.finish().unwrap(); - source.finish().unwrap(); + catalog.close().unwrap(); + source.close().unwrap(); } #[tokio::test] @@ -1776,7 +1776,7 @@ async fn announce_and_unannounce_toggles_discovery() { .expect("timed out requesting the reannounced broadcast") .expect("a reannounced broadcast resolves"); - broadcast.finish().unwrap(); + broadcast.close().unwrap(); } #[tokio::test] @@ -1793,7 +1793,7 @@ async fn finish_unpublishes() { // A graceful finish detaches immediately; the path stops resolving. Removal is // asynchronous, so poll until it takes effect. - broadcast.finish().unwrap(); + broadcast.close().unwrap(); let removed = tokio::time::timeout(TIMEOUT, async { loop { if consumer.request_broadcast("live".into()).await.is_err() { @@ -1862,7 +1862,7 @@ async fn local_publish_consume_audio() { assert_eq!(frame.payload, payload); assert_eq!(frame.timestamp_us, 1_000_000); - broadcast.finish().unwrap(); + broadcast.close().unwrap(); } #[tokio::test] @@ -1924,7 +1924,7 @@ async fn video_publish_consume() { assert_eq!(frame.timestamp_us, 0); assert!(!frame.payload.is_empty(), "frame should have payload data"); - broadcast.finish().unwrap(); + broadcast.close().unwrap(); } /// The raw-video publish path: hand mid-gray RGBA to `encode_video` and check @@ -2037,7 +2037,7 @@ async fn video_raw_publish_consume() { assert!(!frame.payload.is_empty(), "frame should carry encoded video"); video.finish().unwrap(); - broadcast.finish().unwrap(); + broadcast.close().unwrap(); } /// The decode side picks its CPU pixel layout: an unset `format` delivers I420, @@ -2145,7 +2145,7 @@ async fn video_decode_format() { i420.cancel(); rgba_out.cancel(); video.finish().unwrap(); - broadcast.finish().unwrap(); + broadcast.close().unwrap(); } /// Regression: a `MoqVideoProducer` is shared, so its calls land on whichever @@ -2231,7 +2231,7 @@ async fn video_raw_publish_from_many_threads() { let closer = video.clone(); std::thread::spawn(move || closer.finish()).join().unwrap().unwrap(); - broadcast.finish().unwrap(); + broadcast.close().unwrap(); } /// A raw video producer rejects a buffer that isn't one picture at the @@ -2295,7 +2295,7 @@ async fn video_raw_publish_rejects_bad_frames() { Err(MoqError::Closed) )); - broadcast.finish().unwrap(); + broadcast.close().unwrap(); } #[tokio::test] @@ -2350,7 +2350,7 @@ async fn multiple_frames_ordering() { assert_eq!(frame.payload, expected.as_bytes(), "frame {i} has wrong payload"); } - broadcast.finish().unwrap(); + broadcast.close().unwrap(); } #[tokio::test] @@ -2392,17 +2392,22 @@ async fn catalog_update_on_new_track() { assert_eq!(catalog2.audio["0.opus"].label.as_deref(), Some("English")); assert_eq!(catalog2.audio["1.opus"].label, None); - broadcast.finish().unwrap(); + broadcast.close().unwrap(); } #[test] -fn finish_closes_producer() { +fn close_twice_is_a_noop() { let broadcast = MoqBroadcastProducer::new().unwrap(); let init = opus_head(); - let _media = broadcast.publish_audio(audio_init(MoqAudioFormat::Opus, init)).unwrap(); - broadcast.finish().unwrap(); + let _media = broadcast + .publish_audio(audio_init(MoqAudioFormat::Opus, init.clone())) + .unwrap(); + broadcast.close().unwrap(); + broadcast.close().unwrap(); - let err = broadcast.finish().unwrap_err(); + let Err(err) = broadcast.publish_audio(audio_init(MoqAudioFormat::Opus, init)) else { + panic!("publishing after close succeeded"); + }; assert!( matches!(err, crate::error::MoqError::Closed), "expected Closed error, got {err}" @@ -2431,7 +2436,7 @@ async fn announced_broadcast() { .unwrap(); // Finish so consumers observe a deliberate end (the canonical end for a // publisher; dropping without finish reads as a failure). - _broadcast.finish().unwrap(); + _broadcast.close().unwrap(); } fn serve(origin: &MoqOriginProducer, prefix: &str) -> Arc { @@ -2484,7 +2489,7 @@ async fn dynamic_broadcast_request() { assert_eq!(frame.timestamp_us, 20_000); track.finish().unwrap(); - served.finish().unwrap(); + served.close().unwrap(); } /// A prefix serves requests beneath it; cancelling the handle @@ -2524,7 +2529,7 @@ async fn dynamic_serves_a_request_under_a_prefix() { crate::error::MoqProtocolKind::Unroutable, ); - served.finish().unwrap(); + served.close().unwrap(); } /// Tearing the origin down ends every handler with `Closed`. A parked request @@ -3266,7 +3271,7 @@ fn without_runtime() { announced.cancel(); client.cancel(); media.finish().unwrap(); - broadcast.finish().unwrap(); + broadcast.close().unwrap(); drop(client); drop(consumer); drop(announcement); @@ -3367,7 +3372,7 @@ async fn server_client_roundtrip() { // Clean up. Exercise `shutdown()` on the client side and the underlying // `cancel(code)` on the server side, so both shutdown paths run. media.finish().unwrap(); - broadcast.finish().unwrap(); + broadcast.close().unwrap(); cs.shutdown(); server_session.cancel(0); server.cancel(); @@ -3439,10 +3444,10 @@ async fn server_client_roundtrip_auto_origin() { .await .expect("timed out waiting for the loopback broadcast") .expect("an auto-origin session should discover its own announcement"); - local_broadcast.finish().unwrap(); + local_broadcast.close().unwrap(); media.finish().unwrap(); - broadcast.finish().unwrap(); + broadcast.close().unwrap(); cs.shutdown(); server_session.cancel(0); server.cancel(); @@ -3648,7 +3653,7 @@ async fn request_per_session_publish_override() { .expect("expected an announcement"); assert_eq!(announcement.prefix(), "override-only"); - broadcast.finish().unwrap(); + broadcast.close().unwrap(); cs.cancel(0); server_session.cancel(0); server.cancel(); @@ -3776,7 +3781,7 @@ async fn client_reconnects_and_resumes_announcements() { .expect("expected an announcement"); assert_eq!(announcement.prefix(), "after-reconnect"); - broadcast.finish().unwrap(); + broadcast.close().unwrap(); cs.cancel(0); server_session.cancel(0); server.cancel(); @@ -4040,7 +4045,7 @@ async fn video_encoder_follows_a_shrinking_grant() { } video.finish().unwrap(); - broadcast.finish().unwrap(); + broadcast.close().unwrap(); } #[cfg(feature = "video")] @@ -4117,7 +4122,7 @@ async fn set_bitrate_caps_a_later_bandwidth_grant() { assert_eq!(video.applied_bitrate(), 1_000_000); video.finish().unwrap(); - broadcast.finish().unwrap(); + broadcast.close().unwrap(); } async fn one_shot_peers() -> (Arc, Arc, Arc) { diff --git a/rs/moq-ffi/uniffi.toml b/rs/moq-ffi/uniffi.toml new file mode 100644 index 0000000000..ddf0c572fe --- /dev/null +++ b/rs/moq-ffi/uniffi.toml @@ -0,0 +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"] diff --git a/rs/moq-gst/src/sink/imp.rs b/rs/moq-gst/src/sink/imp.rs index e07f6c0178..92e12121a6 100644 --- a/rs/moq-gst/src/sink/imp.rs +++ b/rs/moq-gst/src/sink/imp.rs @@ -641,9 +641,8 @@ impl MoqSink { if let Some(err) = failure { gst::warning!(CAT, "finalize on stop: {err:?}"); } - // Finish the broadcast (a deliberate end, so no dropped-without-finish - // warning) before reaping the session task. - state.broadcast.finish(); + // Close the broadcast before reaping the session task. + state.broadcast.close(); state.session.stop(); self.notify_updates(updates); } diff --git a/rs/moq-gst/src/source/imp.rs b/rs/moq-gst/src/source/imp.rs index c7cd09b910..cadaadecdb 100644 --- a/rs/moq-gst/src/source/imp.rs +++ b/rs/moq-gst/src/source/imp.rs @@ -1271,7 +1271,7 @@ mod session_tests { // End the served rendition and the broadcast, which answers for the reserved one. catalog.finish().unwrap(); video.finish().unwrap(); - broadcast.finish(); + broadcast.close(); // No shutdown is sent: the session has to end on the media draining alone. super::RUNTIME diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index 4b5770a10e..1d3cc5df92 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -2293,11 +2293,11 @@ mod tests { // Tear down the publisher mid-group. The track ends abruptly (the cursor drains the // segments it already saw and ends; the still-open live-edge group is NOT finalized, - // since a reset can't vouch that its media is complete), while finishing the broadcast - // ends it cleanly rather than as a failure. (Clean-end finalization of the + // since a reset can't vouch that its media is complete), and the broadcast closes. + // (Clean-end finalization of the // live edge is covered by segments::tests::next_after_walks_finalized_segments.) drop((catalog, media, registration)); - broadcast.finish(); + broadcast.close(); let end = tokio::time::timeout(Duration::from_secs(5), segments.next()) .await diff --git a/rs/moq-hls/src/export/rendition.rs b/rs/moq-hls/src/export/rendition.rs index f1f1a03691..763ffebe00 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/rs/moq-hls/src/export/rendition.rs @@ -139,9 +139,8 @@ impl Media { let mut handle = self.handle.lock().expect("media lock poisoned"); match handle.binding.poll_broadcast(&kio::Waiter::noop()) { Poll::Ready(Ok(broadcast)) if broadcast.is_closed() => { - // The bound sibling ended (a different first hop replaces the publisher with - // Dropped). Origin finish()es that spliced front, so is_finished() cannot - // tell a rival publisher from a clean VOD end. Rows listed for it must not + // The bound sibling ended, and a broadcast end carries no cause, so a rival + // publisher looks the same as a clean VOD end. Rows listed for it must not // be served from the replacement. window.clear(); if let Ok(next) = source.bind(Some(rel)) { diff --git a/rs/moq-hls/src/server/mod.rs b/rs/moq-hls/src/server/mod.rs index 2686362d40..4e15a17b18 100644 --- a/rs/moq-hls/src/server/mod.rs +++ b/rs/moq-hls/src/server/mod.rs @@ -260,8 +260,8 @@ mod tests { let broadcaster = Broadcaster::new(source, Config::default()) .await .expect("catalog broadcast resolves while announced"); - // Finish the publisher so the resolved broadcast (and the broadcaster) reports closed. - producer.finish(); + // Close the publisher so the resolved broadcast (and the broadcaster) reports closed. + producer.close(); settle().await; broadcaster } @@ -311,6 +311,6 @@ mod tests { let cached = server.inner.broadcasters.lock().unwrap().get("live").cloned(); assert!(cached.is_some_and(|cached| Arc::ptr_eq(&cached, &new))); - new_producer.finish(); + new_producer.close(); } } diff --git a/rs/moq-net/benches/origin.rs b/rs/moq-net/benches/origin.rs index ae998dd904..69011d2f13 100644 --- a/rs/moq-net/benches/origin.rs +++ b/rs/moq-net/benches/origin.rs @@ -364,8 +364,8 @@ fn bench_handoff(c: &mut Criterion) { total += started.elapsed(); drop(subscription); - incumbent.finish(); - standby.finish(); + incumbent.close(); + standby.close(); // Wait for the front to close so the next iteration starts a fresh one. resolved.closed().await; } diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 93d0c2d4d6..ff082426a2 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -352,24 +352,6 @@ impl TrackState { } } -/// How the last source for a path detaches, which decides whether the origin closes -/// the broadcast now or holds it open for a replacement. -/// -/// Only the detach that drops the refcount to zero decides, matching the model's rule -/// for several sources at one path (the front's source selection): an earlier owner that vanished -/// does not outvote the last one still on the path. That keeps two advertisements on -/// one session behaving like the same two on separate sessions, where the model sees -/// two independent sources and the last one out decides. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -enum Detach { - /// The peer retracted the path, or we are rolling back an announce we just made. - /// Nothing is coming back, so close it now. - Graceful, - /// The stream carrying the path went away without retracting it. Abort the - /// source so viewers observe the loss as an error rather than a clean end. - Abrupt, -} - struct BroadcastState { // The route announced into our origin for this namespace, post-charge. route: crate::origin::Route, @@ -381,9 +363,8 @@ struct BroadcastState { // active number of PUBLISH_NAMESPACE messages. count: usize, - // One minted source per requested path under the namespace: finish() on a - // deliberate unannounce, dropping (a dying session) aborts them so viewers - // observe the loss as an error. + // One minted source per requested path under the namespace, each closed + // when its guard drops. sources: HashMap, } @@ -746,14 +727,11 @@ where // ends: a clean close, a decode error, or the peer resetting it. Without this // each namespace keeps its refcount and the source never detaches. // - // Abruptly, including on a clean FIN: closing the stream retracts nothing, since - // the protocol has NAMESPACE_DONE for that. Whatever is still live here outlived - // its channel without being withdrawn, so hold the front open for a reconnect. // This is what moq-lite already does, where the equivalent map is a local whose // guards drop. let res = self.run_namespace_entries(&mut stream, &prefix, &peer, &mut live).await; for path in live { - let _ = self.stop_announce(path, Detach::Abrupt); + let _ = self.stop_announce(path); } res } @@ -797,7 +775,7 @@ where // leave subscriptions on a path the peer no longer offers. tracing::debug!(%path, "dropping reflected namespace"); if live.remove(&path) { - let _ = self.stop_announce(path, Detach::Graceful); + let _ = self.stop_announce(path); } continue; }; @@ -827,7 +805,7 @@ where let path = prefix.join(&msg.suffix); tracing::debug!(%path, "namespace_done"); if live.remove(&path) { - let _ = self.stop_announce(path, Detach::Graceful); + let _ = self.stop_announce(path); } } _ => { @@ -962,7 +940,7 @@ where Ok(_) => { if let Err(err) = self.write_ok(&mut stream, request_id).await { // Local rollback, not a peer unannounce: don't count announce bytes. - let _ = self.stop_announce(path, Detach::Graceful); + let _ = self.stop_announce(path); return Err(err); } } @@ -986,15 +964,7 @@ where .await; if attached { - // Ending cleanly IS the retraction here, unlike a NAMESPACE stream: this stream - // carries exactly one advertisement, and withdrawing it is what ends the stream. - // Any other ending left the advertisement standing, so the peer never withdrew - // it and the loss reads as abrupt (an error, not a clean end). - let detach = match res.is_ok() { - true => Detach::Graceful, - false => Detach::Abrupt, - }; - self.stop_announce(path, detach)?; + self.stop_announce(path)?; } res @@ -1106,7 +1076,7 @@ where let Some(advert) = self.route(held.as_ref(), &peer) else { if std::mem::take(attached) { tracing::debug!(%path, "publish_namespace now loops back; detaching"); - let _ = self.stop_announce(path.clone(), Detach::Graceful); + let _ = self.stop_announce(path.clone()); } self.write_ok(stream, msg.request_id).await?; continue; @@ -1375,24 +1345,18 @@ where } } - fn stop_announce(&mut self, path: PathOwned, detach: Detach) -> Result<(), Error> { + /// Release one advertisement of `path`, closing its sources when it was the last. + fn stop_announce(&mut self, path: PathOwned) -> Result<(), Error> { let mut state = self.state.lock(); match state.broadcasts.entry(path.clone()) { Entry::Occupied(mut entry) => { entry.get_mut().count -= 1; if entry.get().count == 0 { - tracing::debug!(route = %self.origin.absolute(&path), ?detach, "unannounced"); - // Dropping the entry retracts the route (its announcement drops). - let removed = entry.remove(); - for (_, source) in removed.sources { - match detach { - Detach::Graceful => source.finish(), - // Dropping the guard aborts the source, so the loss reads - // as an error rather than a clean end. - Detach::Abrupt => {} - } - } + tracing::debug!(route = %self.origin.absolute(&path), "unannounced"); + // Dropping the entry retracts the route (its announcement drops) and + // closes its sources (their guards drop). + entry.remove(); } } Entry::Vacant(_) => return Err(Error::NotFound), @@ -1468,8 +1432,8 @@ where let dynamic = source.dynamic(); request.accept(&source); - // Retain the source so a retraction can finish it. If the route was - // retracted since the accept, finish it here as that retraction would + // Retain the source so a retraction can close it. If the route was + // retracted since the accept, close it here as that retraction would // have, and still serve what it took on: tracks subscribed since carry on. let guard = crate::model::broadcast::SourceGuard::new(source); let retracted = { @@ -1482,9 +1446,7 @@ where None => Some(guard), } }; - if let Some(guard) = retracted { - guard.finish(); - } + drop(retracted); let this = self.clone(); broadcasts.push(async move { @@ -3443,7 +3405,7 @@ mod tests { // Let the SUBSCRIBE go out, then retract the broadcast before any response. settle().await; - producer.finish(); + producer.close(); settle().await; assert!( @@ -4235,7 +4197,7 @@ mod tests { subscriber.start_announce(path.clone(), advert).unwrap(); settle().await; - subscriber.stop_announce(path, Detach::Graceful).unwrap(); + subscriber.stop_announce(path).unwrap(); assert!( routed_now(&consumer, "room/host").is_none(), "an explicit NAMESPACE_DONE must retract the route", @@ -4356,7 +4318,7 @@ mod tests { /// Several advertisements share one refcounted source, so the detach that empties it /// is the one that counts: the broadcast survives the first stop and closes on the - /// last, whatever kind each detach is. + /// last. /// /// That is the model's own rule for several sources at one path (the front's source /// selection), @@ -4377,7 +4339,7 @@ mod tests { settle().await; // One advertisement's stream dies: the other still holds the source. - subscriber.stop_announce(path.clone(), Detach::Abrupt).unwrap(); + subscriber.stop_announce(path.clone()).unwrap(); settle().await; assert!( routed_now(&consumer, "room/host").is_some(), @@ -4385,7 +4347,7 @@ mod tests { ); // The last owner retracts: the broadcast closes with it. - subscriber.stop_announce(path, Detach::Graceful).unwrap(); + subscriber.stop_announce(path).unwrap(); settle().await; assert!( routed_now(&consumer, "room/host").is_none(), @@ -4452,7 +4414,7 @@ mod tests { // One advertisement, so one unannounce detaches it. If the update had bumped the // refcount, this would leave the route stranded. - subscriber.stop_announce(path, Detach::Graceful).unwrap(); + subscriber.stop_announce(path).unwrap(); assert!(routed_now(&consumer, "room/host").is_none()); } @@ -4505,7 +4467,7 @@ mod tests { ); // One advertisement, so one unannounce detaches it. - subscriber.stop_announce(path, Detach::Graceful).unwrap(); + subscriber.stop_announce(path).unwrap(); assert!(routed_now(&consumer, "room/host").is_none()); } @@ -4534,9 +4496,9 @@ mod tests { assert!(routed_now(&consumer, "room/host").is_some()); // Two advertisements, so it takes two unannounces to retract. - subscriber.stop_announce(path.clone(), Detach::Graceful).unwrap(); + subscriber.stop_announce(path.clone()).unwrap(); assert!(routed_now(&consumer, "room/host").is_some()); - subscriber.stop_announce(path, Detach::Graceful).unwrap(); + subscriber.stop_announce(path).unwrap(); assert!(routed_now(&consumer, "room/host").is_none()); } @@ -4563,9 +4525,9 @@ mod tests { assert!(routed_now(&consumer, "room/host").is_some()); // Two advertisements, so it takes two unannounces to retract. - subscriber.stop_announce(path.clone(), Detach::Graceful).unwrap(); + subscriber.stop_announce(path.clone()).unwrap(); assert!(routed_now(&consumer, "room/host").is_some()); - subscriber.stop_announce(path, Detach::Graceful).unwrap(); + subscriber.stop_announce(path).unwrap(); assert!(routed_now(&consumer, "room/host").is_none()); } @@ -4602,7 +4564,7 @@ mod tests { ); // That supersedes the advertisement it repeats, so the old route is retired. - subscriber.stop_announce(path, Detach::Graceful).unwrap(); + subscriber.stop_announce(path).unwrap(); assert!( routed_now(&consumer, "room/host").is_none(), "the superseded route must not stay attached" @@ -5062,7 +5024,7 @@ mod tests { // What the stream's exit path does with whatever it still holds. for path in live { - subscriber.stop_announce(path, Detach::Graceful).unwrap(); + subscriber.stop_announce(path).unwrap(); } assert!(routed_now(&consumer, "room/a").is_none(), "room/a leaked a refcount"); diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 9013f381cc..c85cb19d73 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -2509,9 +2509,8 @@ impl Announced { } fn declined(&mut self, path: PathOwned) { - if let Some(Some(route)) = self.0.insert(path, None) { - route.finish(); - } + // Dropping a replaced route closes its sources. + self.0.insert(path, None); } /// Record an advertisement before deciding what to do with it. @@ -2522,8 +2521,7 @@ impl Announced { /// at each rejection is what stops the next early return from silently freeing a path /// the peer still holds. /// Only valid on a prefix the peer does not already hold, which the caller establishes - /// with [`Self::contains`]. Overwriting an attached route here would drop its source - /// without finishing it, which is [`Self::declined`]'s job. + /// with [`Self::contains`]. Overwriting an attached route is [`Self::declined`]'s job. fn reserve(&mut self, path: PathOwned) { debug_assert!(!self.0.contains_key(&path), "reserved a prefix already advertised"); self.0.insert(path, None); @@ -2534,9 +2532,8 @@ impl Announced { } fn retire(&mut self, path: &PathOwned) { - if let Some(Some(route)) = self.0.remove(path) { - route.finish(); - } + // Dropping the route closes its sources. + self.0.remove(path); } /// Serve queued requests on every attached route: mint a source per requested @@ -2580,8 +2577,7 @@ struct AnnouncedRoute { route: crate::origin::Route, /// Dropping it retracts the route and rejects its queued requests. dynamic: crate::origin::Dynamic, - /// One minted source per requested path, finished on a clean retraction and - /// aborted (via drop) when the session dies. + /// One minted source per requested path, each closed when its guard drops. sources: HashMap, /// Whether the GOAWAY drain already re-priced this route. drained: bool, @@ -2597,14 +2593,6 @@ impl AnnouncedRoute { } } - /// The peer deliberately retracted the route: finish the minted sources so - /// their consumers observe a clean end, and retract the announcement. - fn finish(self) { - for (_, source) in self.sources { - source.finish(); - } - } - /// Update the announced route in place (a restart). fn update(&mut self, route: crate::origin::Route) { self.route = route.clone(); diff --git a/rs/moq-net/src/model/broadcast.rs b/rs/moq-net/src/model/broadcast.rs index 7cdf0618eb..3b7f90306e 100644 --- a/rs/moq-net/src/model/broadcast.rs +++ b/rs/moq-net/src/model/broadcast.rs @@ -2,8 +2,8 @@ //! //! A [Producer] creates tracks on demand: a [Consumer] subscribes by name, and the //! producer either serves a track it already has or is handed a [`track::Request`] to -//! fill. Both handles are refcounted clones of one broadcast, which closes on -//! [`Producer::finish`] or when the last producer drops. +//! fill. Both handles are refcounted clones of one broadcast, which ends on +//! [`Producer::close`] or when the last producer drops. //! //! [Info] is the broadcast's static metadata, fixed for its lifetime. use crate::{cache, stats, track}; @@ -67,7 +67,7 @@ impl Info { /// Consume this [Info] to create a producer that carries its metadata. /// /// Keep the returned [`Producer`] alive for as long as the broadcast should stay - /// available, and end it with [`Producer::finish`]. See the note on [`Producer`]. + /// available, and end it with [`Producer::close`]. See the note on [`Producer`]. pub fn produce(self) -> Producer { Producer::new(self) } @@ -94,12 +94,11 @@ struct BroadcastState { // joined across per-session tracks. `None` for an ordinary broadcast. spliced: Option, - // Set by an explicit `Producer::finish()` or `Producer::abort()` so `Drop` can - // tell a deliberate shutdown apart from a producer dropped by accident. + // Set once the broadcast ends: `Producer::close()`, an abort, or the last + // producer-side handle dropping. Every lookup after it answers `Unroutable`. closing: bool, - // Set only by `Producer::finish()`: the broadcast ended deliberately, as - // opposed to aborting or losing its producer. + // Set only by the deprecated `Producer::finish()`, for `Consumer::is_finished`. finished: bool, // The error passed to `Producer::abort()`, reported by `Consumer::closed`. @@ -185,17 +184,10 @@ impl BroadcastState { /// /// # Lifetime /// -/// **You must keep this producer alive for as long as the broadcast should stay -/// available.** A broadcast lives as long as at least one [`Producer`] exists; -/// children do *not* keep it alive (cloning a [`Consumer`] or holding a -/// [`track::Producer`] does nothing for the broadcast's lifetime). When the last -/// producer goes away every consumer observes [`Error::Dropped`]. -/// -/// End the broadcast with [`Self::finish`] rather than dropping it. Dropping is an -/// easy footgun in garbage-collected bindings (Go, Python, ...), where the handle -/// can be collected the moment it falls out of scope even while you are still -/// publishing, tearing the stream down mid-broadcast. Dropping the last producer -/// without [`Self::finish`] logs a warning. +/// A broadcast lives until [`Self::close`] or until the last [`Producer`] (or +/// [`Dynamic`]) drops, whichever comes first; both end it the same way. Children +/// do *not* keep it alive: cloning a [`Consumer`] or holding a [`track::Producer`] +/// does nothing for the broadcast's lifetime. #[derive(Clone)] pub struct Producer { // Held behind an Arc so each track born from this broadcast can inherit a shared @@ -250,12 +242,11 @@ impl Producer { /// local consumers and peers alike. Call it once the tracks a subscriber needs /// first (a catalog) exist, so the advertisement lands with them in place: /// consumers act on it immediately. The route retracts on [`unannounce`](Self::unannounce), - /// [`finish`](Self::finish), [`abort`](Self::abort), or the last producer - /// dropping. + /// [`close`](Self::close), or the last producer dropping. /// /// Fails with [`Error::Closed`] on a standalone broadcast (one not created - /// through an origin, so there is nothing to announce into) or once the - /// origin's driver has been dropped. + /// through an origin, so there is nothing to announce into), once the broadcast + /// has closed, or once the origin's driver has been dropped. pub fn announce(&self, route: Route) -> Result<(), Error> { let mut announcer = self.alive.announcer.lock(); let announcer = announcer.as_mut().ok_or(Error::Closed)?; @@ -344,7 +335,7 @@ impl Producer { /// /// Subscribers wait on the name until it is accepted, so a reservation the producer /// ends up never filling has to be dropped or rejected. Ending the broadcast - /// ([`Self::finish`] or [`Self::abort`]) resolves whatever is left. + /// resolves whatever is left. pub fn reserve_track(&self, name: impl Into>) -> Result { let request = track::Request::new(self.info.clone(), name).with_stats(self.stats.clone()); self.state.lock().insert_track(request.weak())?; @@ -428,7 +419,11 @@ impl Producer { } } - /// Create a consumer that can subscribe to tracks in this broadcast. + /// Create a consumer of this one publisher's broadcast. + /// + /// A view of this broadcast object, not of its path: a new publisher at the same + /// path is never spliced into it, so it ends when this broadcast does. Go through + /// an origin for a consumer that should not care which publisher serves the path. pub fn consume(&self) -> Consumer { Consumer { info: self.info.clone(), @@ -438,52 +433,28 @@ impl Producer { } } - /// Cleanly finish the broadcast once you are done publishing. - /// - /// Marks the broadcast as deliberately finished so consumers observe a normal - /// end. Prefer this over dropping the producer: an accidental drop (see the note - /// on [`Producer`]) logs a warning, whereas `finish()` is silent. - /// - /// Ends the broadcast outright: consumers observe a normal end immediately and no - /// new tracks are served, whether or not other producer clones are still alive. - /// Existing tracks stay readable so consumers can drain what they already have. + /// End the broadcast for good, whether or not other clones are still alive. /// - /// A name that was reserved, or requested and never taken by a [`Dynamic`], resolves - /// with [`Error::NotFound`]: nothing can fill it now, so its subscribers fail rather - /// than waiting on a [`track::Info`] that is never coming. A request a handler took - /// is left for that handler to answer. + /// Retracts its announcement and local discovery; a later [`Self::announce`] fails + /// with [`Error::Closed`]. Tracks already handed out carry on and end with their + /// own finish or abort. Every later [`Consumer::track`], and every name reserved or + /// still queued for a handler, answers [`Error::Unroutable`]: the same answer an + /// origin gives for a path nobody publishes. A request a [`Dynamic`] already took is + /// left for that handler to answer. /// - /// Borrows rather than consumes, matching [`track::Producer::finish`]. Finishing - /// declares the end, so it must not depend on the caller also surrendering the - /// handle. + /// Dropping the last producer does the same. Closing twice is a no-op. + pub fn close(&self) { + self.alive.close(); + } + + #[doc(hidden)] + #[deprecated(note = "use close(); a broadcast end carries no cause")] pub fn finish(&self) { - { - let mut state = self.state.lock(); - state.closing = true; - state.finished = true; - // A name that was reserved or queued but never served can't arrive now, - // and `Consumer::track` already answers `NotFound` for one asked about after - // this point. Say the same to whoever asked earlier. - state.reject_unserved(Error::NotFound); - } - // Ending the broadcast is what consumers wait on, so signal it here rather - // than leaving it to the last handle drop. - let _ = self.alive.token.close(); - self.alive.retire(); + self.alive.end(true); } - /// Abort the broadcast, ending it for consumers with `err`. - /// - /// Like [`finish`](Self::finish) the end is immediate, whether or not other - /// producer clones are still alive, and existing tracks stay readable so - /// consumers can drain what they already have (an abort does not cascade into - /// the tracks), while a name nothing ever served resolves with `err` the same - /// way [`finish`](Self::finish) resolves it. Unlike a finish, consumers observe - /// `err` from [`Consumer::closed`], so the end reads as a failure rather than a - /// deliberate one. - /// - /// Consumes the producer: an abort is terminal. Errors if the broadcast was - /// already finished or aborted. + #[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(); @@ -507,8 +478,8 @@ impl Producer { } } -/// Ends the broadcast when the last [`Producer`] or [`Dynamic`] drops, closing the -/// liveness channel every [`Consumer`] watches. +/// Ends the broadcast on [`Producer::close`] or when the last [`Producer`] or +/// [`Dynamic`] drops, closing the liveness channel every [`Consumer`] watches. /// /// A refcount rather than a "am I the last one?" check inside `Drop`: that answer is /// a snapshot, and acting on it is exactly what invalidates it. @@ -516,7 +487,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 finish, abort, or the last producer-side handle + // with the broadcast: on close, abort, or the last producer-side handle // dropping. `None` for a standalone broadcast. announcer: kio::Lock>, } @@ -537,6 +508,29 @@ 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) { @@ -549,15 +543,7 @@ impl Alive { impl Drop for Alive { fn drop(&mut self) { - // Warn if the last exit wasn't an explicit finish(), since consumers will - // then see Error::Dropped (classically a GC-collected handle in a language - // binding that tears the stream down mid-publish). - if !self.state.read().closing { - tracing::warn!( - "broadcast::Producer dropped without finish(). Keep the producer alive while publishing, then call finish()." - ); - } - self.retire(); + self.close(); } } @@ -574,36 +560,22 @@ impl Producer { } /// A session-owned handle to a source broadcast created via -/// [`crate::origin::Producer::create_broadcast`]: [`Self::finish`] ends it -/// deliberately, while dropping the guard aborts it as [`Error::Dropped`] (a dead -/// session), so consumers observe the loss as an error. Shared by the lite and -/// IETF subscribers so the drop-vs-finish contract lives in one place. -pub(crate) struct SourceGuard { - // `Option` so `finish` can consume the producer while `Drop` aborts it. - producer: Option, -} +/// [`crate::origin::Producer::create_broadcast`], closing it on drop even while the +/// session's serve machines still hold its [`Dynamic`]. +/// +/// A peer's retraction and a dead session end the source the same way: a broadcast +/// carries no end cause past this hop. Shared by the lite and IETF subscribers. +pub(crate) struct SourceGuard(Producer); impl SourceGuard { pub fn new(producer: Producer) -> Self { - Self { - producer: Some(producer), - } - } - - /// End the source deliberately: the origin detaches it immediately, - /// unannouncing the path if it was the last. - pub fn finish(mut self) { - if let Some(producer) = self.producer.take() { - producer.finish(); - } + Self(producer) } } impl Drop for SourceGuard { fn drop(&mut self) { - if let Some(producer) = self.producer.take() { - let _ = producer.abort(Error::Dropped); - } + self.0.close(); } } @@ -614,6 +586,7 @@ impl Drop for SourceGuard { /// [`track::Request::accept`]s it with a concrete [`track::Info`] or /// [`track::Request::reject`]s it. Dropped when no longer needed; pending requests /// are automatically aborted. +#[derive(Clone)] pub struct Dynamic { info: Arc, // Keeps the broadcast alive while a handler exists (mirrors a producer). @@ -622,31 +595,51 @@ pub struct Dynamic { // Ingress stats scope, applied to the tracks this handler serves. Empty (no-op) // for an untagged broadcast. stats: stats::Scope, + // Declared after `alive` so it drops second: when this was the broadcast's last + // handle, `Alive` has already ended it and answered every queued request + // `Unroutable`, so the handler's own `Dropped` rejection finds nothing left. + _handler: Handler, } -impl Clone for Dynamic { +/// Counts one live [`Dynamic`], rejecting the queued requests when the last one drops. +struct Handler(kio::Shared); + +impl Handler { + fn new(state: kio::Shared) -> Self { + state.lock().requests.add_handler(); + Self(state) + } +} + +impl Clone for Handler { fn clone(&self) -> Self { - // Mirror `new`: count each live handle. Without this, deriving Clone would - // let `Drop` decrement past `new`'s single increment and prematurely flip - // the handler count to zero, causing future `track` calls to return `NotFound`. - self.state.lock().requests.add_handler(); + // Count each live handle, or dropping a clone would flip the handler count to + // zero and future `track` calls would return `NotFound`. + Self::new(self.0.clone()) + } +} - Self { - info: self.info.clone(), - alive: self.alive.clone(), - state: self.state.clone(), - stats: self.stats.clone(), +impl Drop for Handler { + fn drop(&mut self) { + // Decrement and reject under one lock, so a `track` call that saw a live + // handler through the same lock can't slip a request past the rejection. + let mut state = self.0.lock(); + if state.requests.remove_handler() { + // No handlers left to fulfill pending requests; reject them so consumers + // don't block forever on tracks nobody will serve. + for request in state.requests.drain_queued() { + request.reject(Error::Dropped); + } } } } impl Dynamic { fn new(info: Arc, alive: Arc, state: kio::Shared, stats: stats::Scope) -> Self { - state.lock().requests.add_handler(); - Self { info, alive, + _handler: Handler::new(state.clone()), state, stats, } @@ -659,9 +652,8 @@ impl Dynamic { /// Poll for the next consumer-requested track, without blocking. /// - /// Returns [`Error::Closed`] once the broadcast was deliberately ended - /// ([`Producer::finish`] or aborted), so a serving loop knows to stop and - /// release its handle. + /// Returns [`Error::Closed`] once the broadcast has ended, so a serving loop + /// knows to stop and release its handle. pub fn poll_requested_track(&mut self, waiter: &kio::Waiter) -> Poll> { let mut state = ready!(self.state.poll(waiter, |state| { if state.requests.has_queued() || state.closing { @@ -699,15 +691,14 @@ impl Dynamic { } } - /// Block until the broadcast is closed, by [`Producer::finish`], - /// [`Producer::abort`], or every producer dropping, returning the cause. + /// 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 { kio::wait(|waiter| self.poll_closed(waiter)).await } - /// Poll until the broadcast closes; ready with the cause: the error passed to - /// [`Producer::abort`], or [`Error::Dropped`] for a [`Producer::finish`] or a - /// dropped producer (check [`Consumer::is_finished`] to tell those apart). + /// 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)) @@ -719,21 +710,6 @@ impl Dynamic { } } -impl Drop for Dynamic { - fn drop(&mut self) { - // Decrement and reject under one lock, so a `track` call that saw a live - // handler through the same lock can't slip a request past the rejection. - let mut state = self.state.lock(); - if state.requests.remove_handler() { - // No handlers left to fulfill pending requests; reject them so consumers - // don't block forever on tracks nobody will serve. - for request in state.requests.drain_queued() { - request.reject(Error::Dropped); - } - } - } -} - #[cfg(test)] use futures::FutureExt; @@ -753,6 +729,10 @@ impl Dynamic { } /// Subscribe to arbitrary broadcast/tracks. +/// +/// Its close signal means this broadcast object ended, not that the path went +/// offline: announcements say whether a path is live, and a new publisher may +/// announce the same path again. pub struct Consumer { info: Arc, // Broadcast liveness (read-only): watched for close. @@ -806,6 +786,8 @@ impl Consumer { } /// Get a handle to a track on this broadcast. + /// + /// Fails with [`Error::Unroutable`] once the broadcast has ended. pub fn track(&self, name: &str) -> Result { // Rebind the track to *this* handle's view of the broadcast, so a catalog track // resolves its relative references against the path we were handed out at rather @@ -816,16 +798,17 @@ impl Consumer { } fn track_inner(&self, name: &str) -> Result { - // A closed broadcast (every producer and handler gone) serves nothing. - if self.is_closed() { - return Err(Error::Dropped); - } - let mut state = self.state.lock(); + // An ended broadcast serves nothing new, not even a track it still has: + // the lookup answers what a fresh `request_broadcast` for the path would. + // Tracks already handed out are untouched. + if state.closing { + return Err(Error::Unroutable); + } + // A route-fed broadcast mints spliced logical tracks: they outlive any // session, and a route is asked (via the pending queue) to start serving. - let closing = state.closing; if let Some(spliced) = state.spliced.as_mut() { // An aborted logical track is a verdict from the sources attached at // the time, not a property of the name: a publisher that had not yet @@ -850,11 +833,6 @@ impl Consumer { producer.consume(), )); } - // A deliberately-ended broadcast serves nothing new; nothing drains the - // pending queue once the front is torn down. - if closing { - return Err(Error::NotFound); - } let name: Arc = name.into(); let producer = super::resume::Producer::new(); let consumer = producer.consume(); @@ -882,12 +860,6 @@ impl Consumer { return Ok(pending.consume()); } - // A deliberately-ended broadcast serves nothing new; existing tracks above - // stay readable so consumers can drain the cache. - if state.closing { - return Err(Error::NotFound); - } - // Allocate the name once and share the same Arc across the request, the // requests map, and the FIFO order. The request inherits the broadcast's // cache pool through its `Arc`, same as a producer-created track. @@ -919,32 +891,28 @@ impl Consumer { } } - /// Block until the broadcast is closed, by [`Producer::finish`], - /// [`Producer::abort`], or every producer dropping, and return the cause. + /// Block until the broadcast ends, by [`Producer::close`] or every producer dropping. /// - /// Returns the error passed to [`Producer::abort`], or [`Error::Dropped`] for a - /// [`Producer::finish`] or a dropped producer (check [`Self::is_finished`] to - /// tell those apart). + /// 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) } - /// Returns true if every [`Producer`] has been dropped. + /// Returns true once the broadcast has ended. pub fn is_closed(&self) -> bool { self.alive.is_closed() } - /// Whether the broadcast is on its way out: deliberately ended (finish/abort - /// marked, even while handles remain) or already fully closed. The origin's + /// Whether the broadcast has ended, observed under the state lock. The origin's /// dispatcher treats a rejection from such a source as imminent detach rather /// than a strike. pub(crate) fn is_closing(&self) -> bool { - self.is_closed() || self.state.read().closing + self.state.read().closing } - /// Whether the broadcast ended via a deliberate [`Producer::finish`], as opposed - /// to aborting or losing its producer. `false` while the broadcast is still live. + #[doc(hidden)] + #[deprecated(note = "a broadcast end carries no cause")] pub fn is_finished(&self) -> bool { self.state.read().finished } @@ -1188,7 +1156,7 @@ mod test { assert!(!demand.is_used()); // Every producer gone: both edges report the closure. - producer.finish(); + producer.close(); assert!(matches!(demand.used().await, Err(Error::Dropped))); assert!(matches!(demand.unused().await, Err(Error::Dropped))); } @@ -1305,31 +1273,47 @@ mod test { track1c.assert_not_closed(); } - /// `closed()` reports the cause: the abort error, or `Dropped` for a finish or - /// a dropped producer, with `is_finished` telling the latter two apart. + /// `close()` ends the broadcast for every clone at once, and a second close is a no-op. #[tokio::test] - async fn closed_cause() { - // Abort: the error comes through, and it isn't a finish. + async fn close_ends_every_clone() { + let producer = Info::new().produce(); + let clone = producer.clone(); + let consumer = producer.consume(); + + producer.close(); + assert!(matches!(consumer.closed().await, Error::Dropped)); + assert!(matches!(consumer.track("video"), Err(Error::Unroutable))); + assert!(matches!(clone.consume().track("video"), Err(Error::Unroutable))); + + clone.close(); + producer.close(); + } + + /// Dropping the last producer ends the broadcast exactly like `close()`. + #[tokio::test] + async fn drop_ends_like_close() { + let producer = Info::new().produce(); + let consumer = producer.consume(); + drop(producer); + assert!(matches!(consumer.closed().await, Error::Dropped)); + 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()); - // Finish: a deliberate clean end. let producer = Info::new().produce(); let consumer = producer.consume(); producer.finish(); assert!(matches!(consumer.closed().await, Error::Dropped)); assert!(consumer.is_finished()); - - // Plain drop: neither aborted nor finished. - let producer = Info::new().produce(); - let consumer = producer.consume(); - // Deliberate for the test: exercises the accidental-drop path (warns). - drop(producer); - assert!(matches!(consumer.closed().await, Error::Dropped)); - assert!(!consumer.is_finished()); } #[tokio::test] @@ -1503,24 +1487,25 @@ mod test { } /// A reserved name nobody accepts is the parking case a publisher has to be able to - /// end. Ending the broadcast is where it does: `Consumer::track` already answers - /// `NotFound` for a name asked about after this point, so whoever asked earlier gets + /// end. Ending the broadcast is where it does: `Consumer::track` answers + /// `Unroutable` for a name asked about after this point, so whoever asked earlier gets /// the same answer instead of waiting on info that can never arrive. #[tokio::test] - async fn finish_resolves_a_reserved_name() { + async fn close_resolves_a_reserved_name() { let producer = Info::new().produce(); let consumer = producer.consume(); let _request = producer.reserve_track("track1").unwrap(); let pending = subscribe_pending!(consumer, "track1"); - producer.finish(); - assert!(matches!(pending.await, Err(Error::NotFound))); + producer.close(); + assert!(matches!(pending.await, Err(Error::Unroutable))); } - /// An abort says why the broadcast ended, and an unserved name resolves with that - /// reason rather than a generic failure. + /// 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(); @@ -1539,23 +1524,51 @@ mod test { /// 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] - async fn finish_resolves_a_queued_request() { + async fn close_resolves_a_queued_request() { let producer = Info::new().produce(); let dynamic = producer.dynamic(); let consumer = dynamic.consume(); let pending = subscribe_pending!(consumer, "track1"); - producer.finish(); - assert!(matches!(pending.await, Err(Error::NotFound))); + producer.close(); + assert!(matches!(pending.await, Err(Error::Unroutable))); + drop(dynamic); + } + + /// A queued request answers the same when the broadcast ends by its last handle + /// dropping, even when that handle is the `Dynamic` that would have served it. + #[tokio::test] + async fn dropping_the_last_handle_resolves_a_queued_request() { + let dynamic = Info::new().produce().dynamic(); + let consumer = dynamic.consume(); + + let pending = subscribe_pending!(consumer, "track1"); + drop(dynamic); + assert!(matches!(pending.await, Err(Error::Unroutable))); + } + + /// With a producer still alive, losing the last handler is not the broadcast ending: + /// the queued request fails as `Dropped`. + #[tokio::test] + async fn dropping_the_last_handler_resolves_a_queued_request_dropped() { + let producer = Info::new().produce(); + let dynamic = producer.dynamic(); + let consumer = dynamic.consume(); + + let pending = subscribe_pending!(consumer, "track1"); + + drop(dynamic); + assert!(matches!(pending.await, Err(Error::Dropped))); + producer.close(); } /// A request a handler already took is the handler's to answer: it may be in flight /// to a peer, and a retraction does not disturb subscriptions already in flight. /// Whatever the handler decides still reaches the consumer. #[tokio::test] - async fn finish_leaves_a_claimed_request_to_its_handler() { + async fn close_leaves_a_claimed_request_to_its_handler() { let producer = Info::new().produce(); let mut dynamic = producer.dynamic(); let consumer = dynamic.consume(); @@ -1565,14 +1578,14 @@ mod test { let dropped = subscribe_pending!(consumer, "track2"); let abandoned = dynamic.requested_track().await.unwrap(); - producer.finish(); + producer.close(); assert!( accepted.poll_ok(&kio::Waiter::noop()).is_pending(), - "finish rejected a claimed request" + "close rejected a claimed request" ); assert!( dropped.poll_ok(&kio::Waiter::noop()).is_pending(), - "finish rejected a claimed request" + "close rejected a claimed request" ); let _track = request.accept(None); @@ -1583,11 +1596,11 @@ mod test { } /// A reverse fetch can install the track metadata before the live request is - /// accepted, but it does not create a live publisher. Finishing the broadcast + /// accepted, but it does not create a live publisher. Closing the broadcast /// must still reject that name so an arrival-order subscriber does not park on /// backfill that is deliberately absent from its queue. #[tokio::test] - async fn finish_resolves_an_unaccepted_track_with_fetched_info() { + async fn close_resolves_an_unaccepted_track_with_fetched_info() { let producer = Info::new().produce(); let consumer = producer.consume(); @@ -1601,24 +1614,25 @@ mod test { pending_fetch.await.unwrap(); let mut subscriber = track.subscribe(None).await.unwrap(); - producer.finish(); - assert!(matches!(subscriber.recv_group().await, Err(Error::NotFound))); + producer.close(); + assert!(matches!(subscriber.recv_group().await, Err(Error::Unroutable))); let stale = request.accept(None); assert!(stale.append_group().is_err()); } /// Ending the broadcast doesn't cascade into a track someone is publishing: it keeps - /// its cache and its publisher decides when it ends. + /// its cache and its publisher decides when it ends. Only a new lookup is refused. #[tokio::test] - async fn finish_spares_a_served_track() { + async fn close_spares_a_served_track() { let producer = Info::new().produce(); let consumer = producer.consume(); let track = producer.create_track("track1", None).unwrap(); let mut subscriber = consumer.track("track1").unwrap().subscribe(None).await.unwrap(); - producer.finish(); + producer.close(); + assert!(matches!(consumer.track("track1"), Err(Error::Unroutable))); track.append_group().unwrap(); subscriber.assert_group(); @@ -1627,22 +1641,22 @@ mod test { /// The publisher may still be holding the `track::Request` for a name the broadcast /// just gave up on. Accepting it afterwards must not resurrect the track, or a - /// subscriber that was told `NotFound` could be contradicted by a later one. + /// subscriber that was told `Unroutable` could be contradicted by a later one. #[tokio::test] - async fn finish_leaves_a_stale_reservation_inert() { + async fn close_leaves_a_stale_reservation_inert() { let producer = Info::new().produce(); let consumer = producer.consume(); let request = producer.reserve_track("track1").unwrap(); let pending = subscribe_pending!(consumer, "track1"); - producer.finish(); - assert!(matches!(pending.await, Err(Error::NotFound))); + producer.close(); + assert!(matches!(pending.await, Err(Error::Unroutable))); let track = request.accept(None); assert!(track.append_group().is_err()); let mut subscriber = track.subscribe(None); - assert!(matches!(subscriber.recv_group().await, Err(Error::NotFound))); + assert!(matches!(subscriber.recv_group().await, Err(Error::Unroutable))); assert!(consumer.track("track1").is_err()); } @@ -1659,7 +1673,7 @@ mod test { drop(request); assert!(matches!(pending.await, Err(Error::Dropped))); - producer.finish(); + producer.close(); } /// `track::Request::reject` carries its reason the same way, which is what lets a @@ -1674,7 +1688,7 @@ mod test { request.reject(Error::NotFound); assert!(matches!(pending.await, Err(Error::NotFound))); - producer.finish(); + producer.close(); } /// The interleave every unused-driven teardown has to survive: a wire subscriber @@ -1711,7 +1725,7 @@ mod test { assert!(track.abort_unused(Error::Cancel).is_ok()); assert!(matches!(consumer.track("video"), Err(Error::NotFound))); - producer.finish(); + producer.close(); } #[test] @@ -1724,6 +1738,6 @@ mod test { track.clone().abort(Error::Cancel).unwrap(); assert!(!track.is_used()); assert!(track.abort_unused(Error::Cancel).is_ok()); - producer.finish(); + producer.close(); } } diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 52c06cc5f6..ae1c473d21 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -1237,11 +1237,8 @@ impl Producer { /// polled. Register a [`broadcast::Producer::dynamic`] handler before /// announcing, so the first consumer finds the tracks it serves. /// - /// End the broadcast with [`broadcast::Producer::finish`]; dropping it - /// without finishing also works, but logs a warning. Either way the path - /// closes once it was the last source; an unfinished drop additionally aborts - /// the spliced tracks with an error, so consumers observe a failure rather - /// than a clean end. + /// End the broadcast with [`broadcast::Producer::close`] or by dropping it; + /// either way the path closes once it was the last source. /// /// Fails with [`Error::Unauthorized`] if `path` is outside the prefixes this /// producer may publish under (after [`scope`](Self::scope)), @@ -1271,7 +1268,7 @@ impl Producer { // The broadcast is a route table entry at its exact path from the start, // hidden from cursors and requests until it announces. The entry lives - // as long as the broadcast: its announcer drops on finish, abort, or the + // as long as the broadcast: its announcer drops on close, abort, or the // last handle. let announcing = Announcing { hop: self.hop, @@ -2286,7 +2283,7 @@ async fn run_front(task: FrontTask) { // subscriptions already in flight): dropping their producers // leaves each reader on the copy it was spliced from, ending // when and as that copy ends. - broadcast.finish(); + broadcast.close(); broadcast.release_spliced(err.clone()); for (_, mut io) in tracks.drain() { // A reader still waiting on its source's answer is in flight @@ -4192,7 +4189,7 @@ mod tests { broadcast.announce(Route::default()).unwrap(); announced.assert_next_active("room/alice"); peer.assert_next_active("room/alice"); - broadcast.finish(); + broadcast.close(); announced.assert_next_ended("room/alice"); peer.assert_next_ended("room/alice"); assert!(matches!(broadcast.announce(Route::default()), Err(Error::Closed))); @@ -6123,9 +6120,9 @@ mod tests { assert!(again.is_clone(&resolved)); // Losing one source keeps the front alive; losing both closes it. - first.finish(); + first.close(); settle(|| consumer.get_broadcast("room/alice").is_some()).await; - second.finish(); + second.close(); settle(|| consumer.get_broadcast("room/alice").is_none()).await; // The path is free again for a fresh broadcast. @@ -6161,7 +6158,7 @@ mod tests { group.finish().unwrap(); track.finish().unwrap(); drop(track); - broadcast.finish(); + broadcast.close(); let mut group = next_group(&mut subscription) .await @@ -6206,7 +6203,7 @@ mod tests { // ANNOUNCE_END overtakes the track's end: the route is retracted, and the front // has acted on it, before the track's last group and end arrive. - source.finish(); + source.close(); drop(server); settle(|| resolved.is_closed()).await; let mut group = track.append_group().unwrap(); @@ -6508,7 +6505,7 @@ mod tests { // The incumbent leaving exhausts the table: the refusal is never retried. drop(track); - first.finish(); + first.close(); assert!(matches!(subscription.recv_group().await, Err(Error::Unsupported))); } @@ -6707,7 +6704,7 @@ mod tests { // Publishing through the nested view lands where the root says. let broadcast = nested.create_broadcast("room/chat/live").unwrap(); assert!(producer.consume().get_broadcast("room/chat/live").is_some()); - broadcast.finish(); + broadcast.close(); } #[test] diff --git a/rs/moq-net/tests/broadcast_close.rs b/rs/moq-net/tests/broadcast_close.rs new file mode 100644 index 0000000000..584f7f77b0 --- /dev/null +++ b/rs/moq-net/tests/broadcast_close.rs @@ -0,0 +1,72 @@ +//! `broadcast::Producer::close()` ends a broadcast the same way for a local consumer +//! and a remote one: the handle closes, every new track lookup answers +//! [`Error::Unroutable`](moq_net::Error::Unroutable), a fresh request for the path +//! answers the same, and the broadcast can never be announced again. + +mod support; + +use std::time::Duration; + +use moq_net::{Error, Hop, Version}; +use support::harness::{MockConnectOptions, connect_mock}; + +const TIMEOUT: Duration = Duration::from_secs(10); + +fn produce_origin(hop: u64) -> moq_net::origin::Producer { + let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::new(Hop::new(hop).unwrap())); + tokio::spawn(support::harness::run(driver)); + producer +} + +/// Close a published broadcast and check what `reader`'s origin answers afterwards. +async fn close_then_lookup(publisher: moq_net::origin::Producer, reader: moq_net::origin::Producer) { + let broadcast = publisher.create_broadcast("bcast").unwrap(); + let _track = broadcast.create_track("video", None).unwrap(); + broadcast.announce(Default::default()).unwrap(); + + let consumer = reader.consume(); + tokio::time::timeout(TIMEOUT, consumer.routed("bcast")) + .await + .expect("announce timeout") + .expect("routed"); + let handle = tokio::time::timeout(TIMEOUT, consumer.request_broadcast("bcast")) + .await + .expect("resolve timeout") + .expect("broadcast resolves"); + handle.track("video").expect("a live broadcast serves its track"); + + broadcast.close(); + + tokio::time::timeout(TIMEOUT, handle.closed()) + .await + .expect("the handle never closed"); + assert!(matches!(handle.track("video"), Err(Error::Unroutable))); + assert!(matches!(handle.track("audio"), Err(Error::Unroutable))); + assert!(matches!( + consumer.request_broadcast("bcast").await, + Err(Error::Unroutable) + )); + assert!(matches!(broadcast.announce(Default::default()), Err(Error::Closed))); +} + +#[tokio::test] +async fn close_ends_a_local_consumer() { + tokio::time::pause(); + let origin = produce_origin(1); + close_then_lookup(origin.clone(), origin).await; +} + +#[tokio::test] +async fn close_ends_a_remote_consumer() { + tokio::time::pause(); + for version in ["moq-lite-05", "moq-transport-14", "moq-transport-19"] { + let publisher = produce_origin(1); + let subscriber = produce_origin(2); + let mut options = MockConnectOptions::new(version.parse::().unwrap()); + options.server_publish = Some(publisher.clone()); + options.client_subscribe = Some(subscriber.clone()); + let _pair = connect_mock(options).await; + + close_then_lookup(publisher, subscriber).await; + } +} diff --git a/rs/moq-net/tests/finished_broadcast_mock.rs b/rs/moq-net/tests/finished_broadcast_mock.rs index 30fcefe404..dc64c0506c 100644 --- a/rs/moq-net/tests/finished_broadcast_mock.rs +++ b/rs/moq-net/tests/finished_broadcast_mock.rs @@ -90,7 +90,7 @@ async fn round(finish_broadcast: bool) -> (Vec>, Option) track.finish().unwrap(); drop(track); if finish_broadcast { - broadcast.finish(); + broadcast.close(); } // The session and both origins stay up until the reader is done. diff --git a/rs/moq-relay/src/cluster.rs b/rs/moq-relay/src/cluster.rs index b9ba56453f..a70de866bf 100644 --- a/rs/moq-relay/src/cluster.rs +++ b/rs/moq-relay/src/cluster.rs @@ -1555,9 +1555,9 @@ impl Cluster { } } - // Deliberate shutdown: finishing the registration retracts the route. + // Deliberate shutdown: closing the registration retracts the route. if let Some(registration) = self_registration.as_mut() { - registration.finish(); + registration.close(); } Ok(()) } diff --git a/rs/moq-rtc/src/server/mod.rs b/rs/moq-rtc/src/server/mod.rs index 18e0df6831..580b22f671 100644 --- a/rs/moq-rtc/src/server/mod.rs +++ b/rs/moq-rtc/src/server/mod.rs @@ -88,10 +88,10 @@ impl AcceptedSession { } _ = cancel => { tracing::debug!(role = self.role, "webrtc session terminated by DELETE"); - // A deliberate end: finish the broadcast so the origin + // A deliberate end: close the broadcast so the origin // unannounces it immediately. if let Some(broadcast) = self.broadcast.take() { - broadcast.finish(); + broadcast.close(); } Ok(()) } diff --git a/rs/moq-rtmp/src/dial.rs b/rs/moq-rtmp/src/dial.rs index 4e987e63e1..1a0572b5ad 100644 --- a/rs/moq-rtmp/src/dial.rs +++ b/rs/moq-rtmp/src/dial.rs @@ -485,11 +485,11 @@ async fn client_handshake(stream: &mut S) -> anyhow::Result> /// An active pull: the moq-mux FLV importer publishing into the origin. Mirrors the /// server's publisher; either [`Self::finish`] or dropping it unannounces the -/// path, the former without the dropped-without-finish warning. +/// path. struct Publisher { importer: FlvImport, - // A clone of the importer's producer, so a deliberate end can finish() the - // broadcast (prompt unannounce) even though the importer owns it. + // A clone of the importer's producer, so an end can close the broadcast + // (prompt unannounce) even though the importer owns it. broadcast: moq_net::broadcast::Producer, } @@ -522,16 +522,17 @@ impl Publisher { fn finish(&mut self) -> anyhow::Result<()> { self.importer.finish()?; - self.broadcast.finish(); + self.broadcast.close(); Ok(()) } /// Abort the published tracks with `err` so subscribers see the real cause /// (the remote dropped, a protocol error) rather than a generic `Error::Dropped`. /// - /// Consumes the publisher: the broadcast is done. + /// Consumes the publisher and closes the broadcast. fn abort(self, err: moq_net::Error) { self.importer.abort(err); + self.broadcast.close(); } } diff --git a/rs/moq-rtmp/src/server.rs b/rs/moq-rtmp/src/server.rs index 9087b93a4b..a4ed3c0094 100644 --- a/rs/moq-rtmp/src/server.rs +++ b/rs/moq-rtmp/src/server.rs @@ -1301,11 +1301,11 @@ async fn run_handshake(stream: &mut S, peer: SocketAddr) -> anyhow::R /// An active publish: the moq-mux FLV importer, which owns the origin-created /// [`BroadcastProducer`](moq_net::broadcast::Producer) it publishes into. /// Either [`Self::finish`] or dropping it closes the broadcast and unannounces -/// the path, the former without the dropped-without-finish warning. +/// the path. struct Publisher { importer: FlvImport, - // A clone of the importer's producer, so a deliberate end can finish() the - // broadcast (prompt unannounce) even though the importer owns it. + // A clone of the importer's producer, so an end can close the broadcast + // (prompt unannounce) even though the importer owns it. broadcast: moq_net::broadcast::Producer, } @@ -1343,7 +1343,7 @@ impl Publisher { /// the broadcast so the origin unannounces it immediately. fn finish(&mut self) -> anyhow::Result<()> { self.importer.finish()?; - self.broadcast.finish(); + self.broadcast.close(); Ok(()) } @@ -1351,9 +1351,10 @@ impl Publisher { /// (the client disconnected, a protocol error) rather than a generic /// `Error::Dropped` from the importer being dropped. /// - /// Consumes the publisher: the broadcast is done. + /// Consumes the publisher and closes the broadcast. fn abort(self, err: moq_net::Error) { self.importer.abort(err); + self.broadcast.close(); } } diff --git a/rs/moq-srt/src/ts.rs b/rs/moq-srt/src/ts.rs index baafbb3ef1..8e6ccf2abc 100644 --- a/rs/moq-srt/src/ts.rs +++ b/rs/moq-srt/src/ts.rs @@ -21,14 +21,14 @@ use crate::Result; /// transport packets and retains any partial trailing packet internally for the /// next call (the same pattern `moq-cli import ... stdin ts` uses against stdin). /// Either [`Self::finish`] or dropping the publisher ends the broadcast and -/// unannounces the path, the former without the dropped-without-finish warning. +/// unannounces the path. pub struct Publisher { // TS carries undecoded elementary streams (SCTE-35, teletext, DVB AC-3, ...) // verbatim, so the importer uses the `mpegts` catalog extension rather than the // media-only `()`, which would route those PIDs to `Stream::Ignored` and drop them. importer: ts::Import, - // A clone of the importer's producer, so a deliberate end can finish() the - // broadcast (prompt unannounce) even though the importer owns it. + // A clone of the importer's producer, so an end can close the broadcast + // (prompt unannounce) even though the importer owns it. broadcast: moq_net::broadcast::Producer, } @@ -65,16 +65,17 @@ impl Publisher { /// the broadcast so the origin unannounces it immediately. pub fn finish(&mut self) -> Result<()> { self.importer.finish().map_err(moq_mux::Error::from)?; - self.broadcast.finish(); + self.broadcast.close(); Ok(()) } /// Abort the published tracks with `err` so subscribers see the real cause /// (the SRT caller dropped, a demux error) rather than a generic `Error::Dropped`. /// - /// Consumes the publisher: the broadcast is done. + /// Consumes the publisher and closes the broadcast. pub fn abort(self, err: moq_net::Error) { self.importer.abort(err); + self.broadcast.close(); } } diff --git a/rs/moq-stats/src/produce.rs b/rs/moq-stats/src/produce.rs index e6d860851a..7cd482b042 100644 --- a/rs/moq-stats/src/produce.rs +++ b/rs/moq-stats/src/produce.rs @@ -852,12 +852,11 @@ impl GroupPublisher { self.sessions.adopt_parked(&self.broadcast, &mut self.requested); } - /// Deliberately end the broadcast: finish every pair, then the broadcast - /// itself, so teardown emits no dropped-without-finish warnings. + /// Deliberately end the broadcast: finish every pair, then close the broadcast. fn finish(mut self) { self.traffic.finish(); self.sessions.finish(); - self.broadcast.finish(); + self.broadcast.close(); } } diff --git a/rs/moq-tokio/examples/chat.rs b/rs/moq-tokio/examples/chat.rs index 2e78dc5881..98241f9e5d 100644 --- a/rs/moq-tokio/examples/chat.rs +++ b/rs/moq-tokio/examples/chat.rs @@ -83,10 +83,8 @@ async fn run_broadcast(origin: moq_net::origin::Producer) -> anyhow::Result<()> // Sleep before exiting and closing the broadcast. tokio::time::sleep(tokio::time::Duration::from_secs(10)).await; - // Cleanly close the broadcast so subscribers see a normal end. Dropping the - // producer works too, but closing is explicit (and dropping it by accident - // while still publishing is a common bug this makes obvious). - broadcast.finish(); + // End the broadcast. Dropping the producer does the same; closing is explicit. + broadcast.close(); Ok(()) } diff --git a/rs/moq-tokio/examples/clock.rs b/rs/moq-tokio/examples/clock.rs index 42912d89c1..c792195209 100644 --- a/rs/moq-tokio/examples/clock.rs +++ b/rs/moq-tokio/examples/clock.rs @@ -84,9 +84,8 @@ async fn main() -> anyhow::Result<()> { _ = clock.run() => Ok(()), }; - // Cleanly close the broadcast on exit so subscribers see a normal end - // rather than Error::Dropped. - broadcast.finish(); + // End the broadcast on exit. + broadcast.close(); result } Command::Subscribe => { diff --git a/rs/moq-tokio/src/origin.rs b/rs/moq-tokio/src/origin.rs index b2911dfba5..ac350fff45 100644 --- a/rs/moq-tokio/src/origin.rs +++ b/rs/moq-tokio/src/origin.rs @@ -39,7 +39,7 @@ mod tests { assert_eq!(update.prefix.as_str(), "cam"); assert!(update.kind.is_active()); - broadcast.finish(); + broadcast.close(); let update = announced.next().await.expect("retraction"); assert!(!update.kind.is_active()); } diff --git a/rs/moq-tokio/tests/broadcast.rs b/rs/moq-tokio/tests/broadcast.rs index d6e56e1723..0bb7cfb16e 100644 --- a/rs/moq-tokio/tests/broadcast.rs +++ b/rs/moq-tokio/tests/broadcast.rs @@ -776,7 +776,7 @@ async fn broadcast_moq_lite_06_announce_lifecycle() { // Unannounce: retracted by announce id on the wire. Dropping the announcement // retracts the route; the broadcast's own end is independent. - second.finish(); + second.close(); let update = next_announce(&mut announcements).await; assert_eq!(update.prefix.as_str(), "second"); assert!(!update.kind.is_active(), "expected retraction"); @@ -791,7 +791,7 @@ async fn broadcast_moq_lite_06_announce_lifecycle() { // Replace the route at "first": retract the original (retiring its announce // id on the wire), then announce the same path again (assigning a fresh id). // Await the retraction first so the events cannot coalesce away. - first.finish(); + first.close(); let update = next_announce(&mut announcements).await; assert_eq!(update.prefix.as_str(), "first"); assert!(!update.kind.is_active(), "expected the replaced retraction"); diff --git a/rs/moq-transcode/src/lib.rs b/rs/moq-transcode/src/lib.rs index a9cda09c0d..4884b0b342 100644 --- a/rs/moq-transcode/src/lib.rs +++ b/rs/moq-transcode/src/lib.rs @@ -190,7 +190,7 @@ impl Transcoder { tasks.shutdown().await; derived.finish()?; - output.finish(); + output.close(); Ok(()) } } diff --git a/swift/Sources/Moq/Broadcast.swift b/swift/Sources/Moq/Broadcast.swift index cf52ddba1b..15a29613c6 100644 --- a/swift/Sources/Moq/Broadcast.swift +++ b/swift/Sources/Moq/Broadcast.swift @@ -344,8 +344,15 @@ public final class BroadcastProducer: Sendable { try ffi.removeCatalogSection(name: name) } - /// Finish the broadcast, finalizing the catalog stream. + /// End the broadcast for good: retract it and serve no new tracks. + /// + /// Tracks already subscribed carry on to their own end. Closing again is a no-op. + public func close() throws { + try ffi.close() + } + + @available(*, deprecated, renamed: "close", message: "A broadcast end carries no cause.") public func finish() throws { - try ffi.finish() + try ffi.close() } } diff --git a/swift/Sources/Moq/Origin.swift b/swift/Sources/Moq/Origin.swift index 6a5f4fe2d8..8c8aec69b6 100644 --- a/swift/Sources/Moq/Origin.swift +++ b/swift/Sources/Moq/Origin.swift @@ -34,10 +34,8 @@ public final class OriginProducer: Sendable { /// /// The broadcast is invisible and unroutable, for this origin's consumers /// and peers alike, until `BroadcastProducer.announce(route:)`. Announce it - /// after populating tracks. `finish()` - /// unpublishes immediately, while releasing the producer without finishing - /// also unpublishes but reads to subscribers as a failure rather than a - /// deliberate end. + /// after populating tracks. `BroadcastProducer.close()` ends it for good; + /// releasing the last handle does the same. public func createBroadcast(path: String) throws -> BroadcastProducer { BroadcastProducer(try ffi.createBroadcast(path: path)) } diff --git a/swift/Tests/MoqTests/SmokeTests.swift b/swift/Tests/MoqTests/SmokeTests.swift index 84fa239c41..38485871e1 100644 --- a/swift/Tests/MoqTests/SmokeTests.swift +++ b/swift/Tests/MoqTests/SmokeTests.swift @@ -135,7 +135,14 @@ final class SmokeTests: XCTestCase { let track = try broadcast.publishTrack(name: "events") XCTAssertEqual(try track.name, "events") try track.finish() - try broadcast.finish() + try broadcast.close() + } + + func testBroadcastCloseTwiceIsNoop() throws { + let broadcast = try BroadcastProducer() + try broadcast.close() + try broadcast.close() + XCTAssertThrowsError(try broadcast.publishTrack(name: "events")) } func testVideoHintsReachMediaPublishApi() throws { @@ -148,13 +155,13 @@ final class SmokeTests: XCTestCase { ) let media = try broadcast.publishVideo(format: .avc3, hint: hint) try media.finish() - try broadcast.finish() + try broadcast.close() } func testVideoPropertiesUseDefaultedFields() throws { let broadcast = try BroadcastProducer() try broadcast.setVideoProperties(VideoProperties(rotation: 315)) - try broadcast.finish() + try broadcast.close() } func testBroadcastConsumerFetchesCachedGroup() async throws { @@ -199,7 +206,7 @@ final class SmokeTests: XCTestCase { consumer.cancel() try producer.finish() - try broadcast.finish() + try broadcast.close() } func testJsonStreamRoundTrip() async throws { @@ -220,7 +227,7 @@ final class SmokeTests: XCTestCase { consumer.cancel() try producer.finish() - try broadcast.finish() + try broadcast.close() } func testJsonProducersReportDemand() async throws { @@ -244,7 +251,7 @@ final class SmokeTests: XCTestCase { try await snapshotDemand.unused() try await streamDemand.unused() - try broadcast.finish() + try broadcast.close() } func testRawTrackTimestamps() async throws { @@ -270,7 +277,7 @@ final class SmokeTests: XCTestCase { XCTAssertEqual(groupFrame?.timestampUs, 23_456) try track.finish() - try broadcast.finish() + try broadcast.close() } func testReadFrameSkipsEmptyThenPopulatedGroups() async throws { @@ -287,7 +294,7 @@ final class SmokeTests: XCTestCase { XCTAssertEqual(frame?.timestampUs, 2_000) try track.finish() - try broadcast.finish() + try broadcast.close() } func testSparseGroupsAndKnownEnd() throws { @@ -301,7 +308,7 @@ final class SmokeTests: XCTestCase { try track.createGroup(sequence: 4).finish() XCTAssertThrowsError(try track.createGroup(sequence: 5)) try track.finish() - try broadcast.finish() + try broadcast.close() } /// `frameDurationUs` is microseconds so Opus' 2.5 ms frame is expressible at @@ -331,7 +338,7 @@ final class SmokeTests: XCTestCase { XCTFail("2 ms is not an opus frame duration: \(error)") } - try broadcast.finish() + try broadcast.close() } /// The decode side picks its CPU layout: an unset `format` is I420, and RGBA @@ -388,7 +395,7 @@ final class SmokeTests: XCTestCase { XCTAssertTrue(stride(from: 3, to: frame.data.count, by: 4).allSatisfy { frame.data[$0] == 0xFF }) try video.finish() - try broadcast.finish() + try broadcast.close() } func testEncodeAudioWithOpusObject() throws { @@ -408,7 +415,7 @@ final class SmokeTests: XCTestCase { try producer.write(silence) XCTAssertEqual(try producer.name, "mic") try producer.finish() - try broadcast.finish() + try broadcast.close() } // Release the config before finishing: the producer retains what it needs. @@ -422,7 +429,7 @@ final class SmokeTests: XCTestCase { } try producer.write(silence) try producer.finish() - try broadcast.finish() + try broadcast.close() } } }