moq-net: split TrackConsumer into a track handle + TrackSubscriber#1576
Merged
Conversation
Separate "a handle to a track you can subscribe to" from "a live
subscription you read from", and move the SUBSCRIBE_OK await off the
broadcast lookup and onto the subscription itself.
- Rename the old read+demand handle `TrackConsumer` -> `TrackSubscriber`.
It keeps everything (recv_group/next_group/read_frame, update,
subscription) and still carries the per-subscriber Subscription that
feeds the producer's aggregate.
- Add a new lightweight `TrackConsumer { broadcast, name }`: a cheap,
cloneable handle that names a track but sends nothing until you call
`subscribe(Subscription).await -> TrackSubscriber` (blocks on
SUBSCRIBE_OK) or `subscribe_pending` for poll-based callers. `fetch`
is reserved (TODO) for one-shot range retrieval.
- `BroadcastConsumer::subscribe_track(name, sub) -> TrackPending` becomes
`consume_track(name) -> TrackConsumer`; the await now lives on
`.subscribe()`. All call sites migrated.
The FFI surface is unchanged: `MoqBroadcastConsumer::subscribe_track`
keeps its name and async signature, so the py/swift/kt/go bindings need
no regeneration. The wire format is unchanged, so `js/net` (which has its
own `subscribe(name, priority)` API) needs no change.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
kixelated
commented
Jun 1, 2026
| let track_consumer = broadcast | ||
| .subscribe_track(&track_name, moq_net::Subscription::new(1)) | ||
| .consume_track(&track_name) | ||
| .subscribe(moq_net::Subscription::new(1)) |
Collaborator
Author
There was a problem hiding this comment.
Can you remove ::new? It's not obvious this is priority.
Force struct construction and/or make a builder. moq_net::Subscription::default().with_priority(1)
Collaborator
Author
There was a problem hiding this comment.
Done in f6e7b71. Replaced Subscription::new(1) (which didn't actually exist) with builder methods: Subscription::default().with_priority(1). Added with_priority/with_ordered/with_stale/with_group_start/with_group_end and updated hang::Catalog::default_subscription too.
(Written by Claude)
Comment on lines
+691
to
+700
| pub async fn subscribe(&self, subscription: Subscription) -> Result<TrackSubscriber, Error> { | ||
| self.subscribe_pending(subscription).ok().await | ||
| } | ||
|
|
||
| /// Like [`Self::subscribe`], but returns the unresolved [`TrackPending`] so a | ||
| /// poll-based caller can drive it via [`TrackPending::poll_ok`]. | ||
| pub fn subscribe_pending(&self, subscription: Subscription) -> TrackPending { | ||
| self.broadcast.request_subscribe(&self.name, subscription) | ||
| } | ||
| } |
Collaborator
Author
There was a problem hiding this comment.
I think we just combine these. TrackPending could impl Future
Collaborator
Author
There was a problem hiding this comment.
Done in f6e7b71. TrackPending now impls Future, so TrackConsumer::subscribe(sub) returns it directly: .await it for the TrackSubscriber, or drive it with poll_ok from a poll loop (moq-mux's container source). Dropped subscribe_pending and the redundant ok().
(Written by Claude)
Address review feedback on #1576: - Combine `TrackConsumer::subscribe` and `subscribe_pending` into one method. `TrackPending` now impls `Future`, so `subscribe(sub)` returns it directly: `.await` it for the `TrackSubscriber`, or drive it with `poll_ok` from a poll loop. Drops the redundant `ok()` method. - Add `Subscription` builder methods (`with_priority`, `with_ordered`, `with_stale`, `with_group_start`, `with_group_end`) so the priority is named at the call site instead of a bare positional arg. Update the native doc and `hang::Catalog::default_subscription` to use them. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
kixelated
added a commit
that referenced
this pull request
Jun 1, 2026
kixelated
added a commit
that referenced
this pull request
Jun 1, 2026
CI caught the Lite05 timestamp tests still using the old `bc.subscribe_track(name, sub).ok().await` form, which #1576 removed from `BroadcastConsumer` in favor of the explicit `consume_track(name).subscribe(sub).await` two-step. Update both Lite05 integration tests to match dev's new pattern. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Picks up the next step after #1540. Separates a handle to a track you can subscribe to from a live subscription you read from, and moves the SUBSCRIBE_OK await off the broadcast lookup and onto the subscription itself.
TrackConsumer→TrackSubscriber. It keeps everything (recv_group/next_group/read_frame,update,subscription) and still carries the per-subscriberSubscriptionthat feeds the producer's aggregate.TrackConsumer { broadcast, name }: a cheap, cloneable handle that names a track but sends nothing until you call it.async fn subscribe(Subscription) -> Result<TrackSubscriber>blocks on SUBSCRIBE_OK.subscribe_pending(Subscription) -> TrackPendingfor poll-based callers (moq-mux's containersource.rs).fetchis reserved (TODO) for one-shot range retrieval; not added yet.BroadcastConsumer::subscribe_track(name, sub) -> TrackPendingbecomesconsume_track(name) -> TrackConsumer; the await now lives on.subscribe():All ~30 call sites across the workspace migrated.
Cross-package
moq-ffi/libmoq): unchanged surface.MoqBroadcastConsumer::subscribe_trackkeeps its name and async signature; only its internalself.inner.consume_track(...)calls were renamed. No py/swift/kt/go regeneration needed.js/net: no change. Wire format is unchanged, and it has its ownsubscribe(name, priority)API.doc/lib/rs/env/native.md. Python doc examples are unchanged (FFI API unchanged).Test plan
cargo check --workspace --all-targetscleancargo clippy --workspace --all-targetsclean (via pinned nix toolchain)cargo testformoq-net(328),moq-native,moq-relay,moq-mux— all passcargo fmtvia nixFollow-ups (not in this PR)
stalefield inaggregate_subscriptionlooks inverted (forwards the soonest skip deadline upstream rather than the most patient subscriber's). Flagged for a separate change once confirmed.(Written by Claude)