caprid_relay.media

From the caprid-relay package: import caprid_relay.media

Media policy over registered tracks and subscriptions: the catalog and document services, the frame writer, the track properties that give a media track its default delivery timeout and publisher priority, the subscription parameters that give a viewer’s subscription to it the matching subscriber priority, and the frame reader.

This is the part of stage 2 in spec/media-layer.md that the Video.js streaming example proved against the relay. A catalog is a state-shaped track whose document is re-sent on every relay subscription, and a document service does the same for a document the application replaces, such as a media timeline; a media track maps encoded frames onto groups (one GOP per group, or one frame per group), skips the frames the relay has no subscription for, resumes at the next keyframe after any failure, drops rather than queues when the transport falls behind, and says when it needs a keyframe. The module knows nothing about LOC or MSF: a frame carries its own object properties and the catalog is bytes. Encoding, pacing and time-based frame dropping stay with the application; stale is how it tells the frame writer that a frame is too old to send.

A failure is one of two things (see CONTEXT.md). A retirement ends the track under the writer: the error code is CLOSED, or the track reports closed afterwards. Everything else is an interruption: the group is reset, the track stays open, and publishing resumes at the next keyframe. The “Media publishing” section of spec/behavior.md is the contract these classes implement.

On the subscribing side, FrameReader reads a subscription or a want as frames in decode order. It drives a sequencer it is given (moq_media.FrameSequencer), which knows the format and the deadlines; the “Media subscribing” section of spec/behavior.md is its contract.

Functions

media_subscribe_parameters

def media_subscribe_parameters(
    kind: MediaKind,
    overrides: MessageParameters | None = None,
) -> MessageParameters

The subscription parameters a viewer gives a media track, so audio goes ahead of video on a congested downlink.

Pass them to subscribe, want or switch_to (spec/behavior.md, “Media subscribing”):

audio = subscriber.subscribe(namespace, "audio", handlers, parameters=media_subscribe_parameters("audio"))
video = subscriber.subscribe(namespace, "video", handlers, parameters=media_subscribe_parameters("video"))

The result is a new MessageParameters with a SUBSCRIBER_PRIORITY of 16 for audio and 96 for video, lower being more urgent: the same values media_track_properties gives as publisher priorities. moq-relay 0.17 orders what it sends a viewer by subscriber priority alone and ignores the publisher’s, so this is the priority that decides delivery there. It cannot be changed later: on that relay any update() ends the subscription. overrides are copied onto the result, such as a location_filter; an explicit subscriber_priority, an integer from 0 to 255, replaces the default.

Source

media_track_properties

def media_track_properties(
    kind: MediaKind,
    *,
    keyframe_interval_us: int | None = None,
    target_latency_us: int | None = None,
    subgroup_timeout_us: int | None = None,
    publisher_priority: int | None = None,
) -> list[KeyValuePair]

The track properties that give a media track its default delivery timeout and publisher priority.

Pass them to register_track with the track’s format properties, so a stall costs stale frames rather than lasting latency:

video = publisher.register_track(
    "video",
    track_properties=[
        *loc_track_properties(),
        *media_track_properties("video", keyframe_interval_us=1_000_000),
    ],
)

The result is a SUBGROUP_DELIVERY_TIMEOUT (0x06) in whole milliseconds, rounded up: 500 ms for audio, and for video the target latency less the keyframe interval, but never less than the interval. Every duration is a positive integer number of microseconds. keyframe_interval_us is required for video; target_latency_us (see CONTEXT.md) defaults to twice it and may not be smaller; audio ignores both. subgroup_timeout_us replaces the default for either kind. Then a Default Publisher Priority (0x0E) that puts audio ahead of video: 16 for audio, 96 for video, lower being more urgent; publisher_priority, an integer from 0 to 255, replaces it. moq-relay 0.17 passes it on to viewers in SUBSCRIBE_OK but orders delivery by subscriber priority alone, so on that relay it changes nothing yet. No object timeout is set, in either SDK: through a frame writer it would never fire in a browser, because a group’s wait for settling room is not counted and, once its stream is open, the browser’s transport takes each object at once even during a stall.

Source

Classes

CatalogService

class CatalogService

Serve a state-shaped track: send its document on every relay subscription.

Relay demand is the trigger. demand_changes() yields the current state first and then every new or replacement relay subscription, so a viewer who joins after the relay reconnected gets the document again (the repeated True the vocabulary describes). Viewers who join during one relay subscription trigger nothing and get the relay’s cached copy, so the document must be safe to serve from cache. A write that fails while the track stays open is retried after retry_delay seconds for as long as demand remains, keeping only this document. A failure that ends the track ends the service.

Source

Attributes

• sends: int (property) — Documents the transport accepted.
• failures: int (property) — Writes that failed while the track stayed open, each followed by a retry or by lost demand.

CatalogService.serve

async def serve(self) -> None

Run until cancelled or the track ends.

Returns when the track ends without a write having failed, even during a retry pause. Raises the RelayError that ended the track when a write exposed it, so a retired catalog cannot go unnoticed.

DocumentService

class DocumentService

Serve a state-shaped track whose document the application replaces.

The latest document is sent as a new group on every relay subscription, as CatalogService does, and again on every update while the relay is subscribed. Nothing is sent until there is a document. Updates made while the relay has no subscription only replace the document. A write that fails while the track stays open is retried after retry_delay seconds with the latest document, for as long as demand remains. A failure that ends the track ends the service.

The msf-01 media timeline is this shape: each new entry is a new independent document. One service serves one broadcast; a supervised publisher’s next generation registers a new track and a new service.

Source

Attributes

• document: bytes | None (property) — The latest document, or None before the first.
• sends: int (property) — Documents the transport accepted.
• failures: int (property) — Writes that failed while the track stayed open, each followed by a retry or by lost demand.

DocumentService.update

def update(self, document: bytes) -> None

Replace the document, and send it if the relay is subscribed. Returns at once.

DocumentService.serve

async def serve(self) -> None

Run until cancelled or the track ends.

Returns when the track ends without a write having failed, even during a retry pause. Raises the RelayError that ended the track when a write exposed it, so a retired track cannot go unnoticed.

FrameOutcome

class FrameOutcome

What became of one frame handed to FrameWriter.write.

Source

Attributes

• kind: FrameOutcomeKind
• error: RelayError | None — The failure behind interrupted or ended.
• group_id: int | None — The group the frame was written to, or the one opened or reset on its behalf.

FrameReader

class FrameReader(Generic[F])

Read a track as frames: feed the sequencer every event, and poll it as the application reads, or when a wait it reports ends (at most 50 ms on, since another track may move a shared clock). Iterate it for the frames, in decode order; close ends the iteration.

Pass subscription_handlers to RelaySubscriber.subscribe or want_handlers to SupervisedSubscriber.want. Each group is one subgroup, as the frame writer sends it, so a subgroup’s end is its group’s. A want read through a reader keeps drain at 0, the default: a drained old subscription’s groups would arrive after the Move, indistinguishable from the new one’s. Frames wait in the sequencer until they are read, and lateness is judged then, so a slow reader gets fresh frames rather than a backlog.

Source

Attributes

• closed: bool (property) — Whether close was called.

FrameReader.close

def close(self) -> None

Stop delivering and drop frames not yet read. Later events are ignored.

FrameReader.aiter

def __aiter__(self) -> FrameReader[F]

FrameReader.subscription_handlers

def subscription_handlers(
    self,
    *,
    on_error: Callable[[RequestError], None] | None = None,
    on_publisher_done: Callable[[PublishDone], None] | None = None,
) -> RelaySubscriptionHandlers

Handlers for one subscription, with the application’s own error and completion callbacks.

FrameReader.want_handlers

def want_handlers(
    self,
    *,
    on_subscribed: Callable[[SubscribeOk, tuple[str, ...]], None] | None = None,
    on_publisher_done: Callable[[PublishDone], None] | None = None,
    on_restarted: Callable[[RestartReason, tuple[str, ...]], None] | None = None,
    on_switched: Callable[[WantSwitchedInfo], None] | None = None,
    on_error: Callable[[BaseException], None] | None = None,
) -> WantHandlers

Handlers for a want. A Restart and a Move start the sequencer over; the application’s callbacks run after.

FrameReaderPoll

class FrameReaderPoll(Protocol[F_co])

What a sequencer’s poll() returns: the frames ready now, and when to poll again in microseconds.

Source

Attributes

• frames: Sequence[F_co] (property)
• wait_us: int | None (property)

FrameReaderSequencer

class FrameReaderSequencer(Protocol[F_co])

What a FrameReader drives: moq_media.FrameSequencer qualifies.

Source

Attributes

• track_properties: Sequence[KeyValuePair]

FrameReaderSequencer.object

def object(self, obj: MoqtObject) -> None

FrameReaderSequencer.group_ended

def group_ended(self, group_id: int) -> None

FrameReaderSequencer.group_reset

def group_reset(self, group_id: int) -> None

FrameReaderSequencer.restart

def restart(self) -> None

FrameReaderSequencer.poll

def poll(self, max_frames: int | None = None) -> FrameReaderPoll[F_co]

FrameWriter

class FrameWriter(Generic[F])

Map encoded frames onto a registered track’s groups, one frame at a time.

grouping="keyframe" is the LOC mapping for video: a keyframe FINs the open group and opens a new one, deltas append to it. grouping="frame" sends every frame as its own single-object group, the mapping for audio and any other independently decodable frames.

write may be called again before the previous call returned, from any task: calls are served one at a time in call order, and the frames waiting meanwhile are the writer’s backlog, at most max_backlog of them (default 8 under keyframe grouping, 25 under frame grouping). A frame arriving at a full backlog means the transport fell behind, and frames are dropped rather than queued: under keyframe grouping the backlog and the open group (with the frame being written into it) go, and deltas are dropped until the next keyframe; under frame grouping the oldest waiting frame goes.

write never waits for relay demand. A frame the relay has no subscription for is skipped at once, because an encoder loop cannot block on the write deadline. Every relay demand change resets the open group before the next frame, so publishing resumes at the next keyframe. A frame the caller’s stale predicate rejects is stale, and so are the deltas that depended on a dropped keyframe. A group that fails is reset and the frame is interrupted, with the error; the next keyframe opens a new group. A failure that ends the track (see CONTEXT.md: retirement) is ended, with the error, and every later frame is ended too, as is the first frame after the track closed with nothing in flight.

Under keyframe grouping, keyframe_needed says the next frame the writer can send must be a keyframe, and on_keyframe_needed is called each time it becomes true after the start, so an encoder can force one instead of waiting out its interval. The writer watches the track’s demand from its creation, so it must be created with an event loop running.

Source

Attributes

• closed: bool (property) — True once close ran or the track ended under the writer.
• ended: RelayError | None (property) — The failure that ended the track, once a write has seen it; None while the track is open.
• groups: int (property) — Groups opened so far.
• objects: int (property) — Objects the transport accepted so far.
• backlog: int (property) — Frames waiting behind the one being written.
• keyframe_needed: bool (property) — True while the next frame this writer can send must be a keyframe; always False under frame grouping.

FrameWriter.write

async def write(self, frame: F) -> FrameOutcome

Send one frame if the relay wants it, and say what happened.

Returns when the transport accepted the object, or as soon as the frame is known not to be sent. Raises RelayError with code CLOSED after close; a track that ended on its own is reported as an ended outcome instead, because the encoder keeps producing.

FrameWriter.close

async def close(self) -> None

Drop the backlog, FIN the open group, if any, and refuse further frames. Idempotent.

A frame being written finishes first. A RelayError from the FIN is swallowed: the group was abandoned either way. Anything else propagates. The registered track stays the application’s to close.

ReplaceableTrack

class ReplaceableTrack(Generic[F])

A track the publisher can replace mid-stream, for barge-in.

A voice agent that is talked over replaces its reply track, and listeners drop the old reply’s audio and play the new track. msf-01 §5.3 makes a changed track a new name, so replace registers the next track ({track}-{n + 1} for replacement n from 1, so audio-2 first, or name(n)) and lists it in the catalog in the current track’s place; a listener that sees that (moq_media.find_replacement_track) plays the new track. The old track’s frames that have not been sent are dropped, and the track is closed. The catalog is whatever build_catalog(name) returns for the track that is current, handed to catalog_service.update at creation and at each replacement.

Frames go through a FrameWriter per track, built with writer_options (default grouping="frame", the mapping for audio). A listener subscribes to a replacement only once the new catalog has reached it, about a round trip, and a frame writer skips what it is handed before then: the first words of the new reply. So frames written to a replacement before the relay subscribes wait, in order, and go out once it does; any still waiting after demand_wait seconds go to the writer, which skips them. demand_wait=0 waits for nothing. The first track never waits.

Create one with an event loop running: the frame writers need it. The first track is registered at once, and the publisher’s track name rules apply to every name: they are unique per publisher connection, including after a track closes.

Source

Attributes

• track: str (property) — The current track’s name.
• replacements: int (property) — How many times the track has been replaced: 0 while the first track plays.
• closed: bool (property) — True once close ran.

ReplaceableTrack.write

async def write(self, frame: F) -> FrameOutcome

Send one frame on the current track and say what happened, as FrameWriter.write does.

A frame written to a replacement before the relay has subscribed to it waits (demand_wait). A frame of a track that is replaced before it went out is dropped. Raises RelayError with code CLOSED after close.

ReplaceableTrack.replace

async def replace(self) -> TrackReplacement

Replace the track: register the next track, list it in the catalog, then drop and close the old one.

What the current track has not sent is dropped. Returns once the old track is closed. Frames written from the call on go to the new track. A failure to register the new track or to update the catalog is raised and leaves the current track in place. Raises RelayError with code CLOSED after close.

ReplaceableTrack.close

async def close(self) -> None

Drop what has not been sent and close the current track. Idempotent.

write and replace raise after it.

TrackReplacement

class TrackReplacement

What ReplaceableTrack.replace did.

Source

Attributes

• from_track: str — The track replaced, now closed.
• to_track: str — The new track, listed in the catalog in its place.
• replacements: int — How many times the track has been replaced, this time included: 1 for the first replacement.

Protocols

Frame

class Frame(Protocol)

One encoded frame as the frame writer sees it. Any object with these attributes qualifies.

properties are the object properties sent with the frame, already built by the caller (for LOC media, moq_media.loc_properties). key marks a random-access point: under keyframe grouping it starts a new group. discontinuity says frames before this one were dropped, so an open group can no longer be completed and is reset.

Source

Attributes

• payload: bytes (property)
• key: bool (property)
• properties: Sequence[KeyValuePair] (property)
• discontinuity: bool (property)

Constants

FrameOutcomeKind

FrameOutcomeKind = ...

What became of a frame: sent, or one of the five ways it was not.

Source

MediaKind

MediaKind = Literal['audio', 'video']

What a media track carries, which decides its default delivery timeout and priorities.

Source