moq_core

From the caprid-moq-core package: import moq_core

Transport-independent MOQT track primitives.

The Python counterpart of ts/packages/moq-core: protocol encoding, sessions, and stream handling over an abstract transport, with no runtime dependencies and no knowledge of Relay. MOQT_PROTOCOL_ID is the default draft every SDK shares and MOQT_PROTOCOL_IDS the drafts a client offers; spec/protocol.json is their source of truth and tests/test_spec_protocol.py checks they agree.

Functions

create_moqt_publish_session

def create_moqt_publish_session(
    transport: MoqtTransport,
    config: MoqtPublishSessionConfig | None = None,
) -> MoqtPublishSession

Create a publish-capable MOQT session driver over an established transport. Requires a running event loop.

Example:

session = create_moqt_publish_session(transport)
await session.ready
session.announce(["live", "abc123"])
track = session.register_track(RegisterTrackOptions(track_namespace=["live", "abc123"], track_name="video"))

Source

create_moqt_session

def create_moqt_session(
    transport: MoqtTransport,
    config: MoqtSessionConfig | None = None,
) -> MoqtSession

Create a MOQT session driver over an established transport. Requires a running event loop.

Example:

session = create_moqt_session(transport, MoqtSessionConfig(callbacks=MoqtSessionCallbacks(on_goaway=migrate)))
await session.ready
subscription = session.subscribe(
    MoqtSubscribeOptions(track_namespace=["live", "stream1"], track_name="catalog"),
    SubscriptionHandlers(on_object=handle_catalog_object),
)

Source

create_subgroup_writer

def create_subgroup_writer(
    stream: SendStream,
    *,
    track_alias: int,
    group_id: int,
    priority: int | None = None,
    has_properties: bool = True,
    end_of_group: bool = True,
) -> SubgroupWriter

Open a subgroup writer over an established unidirectional stream.

The SUBGROUP_HEADER is queued immediately; objects follow in write_object call order. priority omitted sets the DEFAULT_PRIORITY flag so subscribers fall back to the subscription’s priority. has_properties makes every object carry a Properties block (an empty list still encodes as a zero length, so mixed-metadata tracks need no special casing). end_of_group declares that the subgroup contains the group’s largest object on FIN.

Source

is_moqt_protocol_error

def is_moqt_protocol_error(value: object) -> bool

True when value is a MoqtProtocolError: the session drivers close the transport on those alone.

Source

Classes

AbsoluteLocationFilter

class AbsoluteLocationFilter

An explicit start, optionally bounded by an end group.

The end group is a delta from the start group and is included whole unless end_object narrows it. Types 0x02 to 0x04 on draft-22, two to four fields on draft-20. On draft-20 an open-ended absolute {0,0} is unfiltered by definition and encodes as NoLocationFilter, since its two-field spelling reads as NextObjectFilter; draft-22 has a spelling for it.

Source

Attributes

• start: Location
• end_group_delta: int | None
• end_object: int | None
• type: Literal['absolute']

AnnounceEnded

class AnnounceEnded

What on_announce_ended receives: no solicitation stream carries the announce any more.

error is a MoqtProtocolError whose __cause__ is the stream failure, if any, that ended the last carrier.

Source

Attributes

• error: BaseException | None

BidirectionalStream

class BidirectionalStream

A bidirectional stream as its two halves.

Source

Attributes

• readable: ReceiveStream
• writable: SendStream

Closed

class Closed

How a session or connection ended. error is None for a clean close.

Source

Attributes

• error: BaseException | None

FetchEndOfRange

class FetchEndOfRange

An End of Range marker on a fetch data stream (§11.4.1): a run of Locations was not serialized.

status says why; group_id and object_id are the Location the marker carries.

Source

Attributes

• kind: Literal['end-of-range']
• status: FetchEndOfRangeStatus
• group_id: int
• object_id: int

FetchHandle

class FetchHandle

A live fetch request; its entries arrive through the FetchHandlers it was given.

Source

Attributes

• request_id: int (property) — The Request ID the FETCH carried (§6.4.2.1).

FetchHandle.cancel

def cancel(self, reason: object = None) -> None

Stop the fetch: its request stream is cancelled and no further callbacks run. Idempotent.

FetchHandlers

class FetchHandlers

Callbacks for one fetch. Every one is optional.

Source

Attributes

• on_ok: Callable[[FetchOk], None] | None — The publisher accepted the fetch.
• on_error: Callable[[RequestError], None] | None — The request failed: a REQUEST_ERROR, or a stream failure or timeout in the same shape.
• on_entry: Callable[[FetchStreamEntry], None] | None — An object or End of Range marker arrived on the fetch data stream.
• on_end: Callable[[], None] | None — The fetch data stream finished cleanly (FIN).
• on_reset: Callable[[FetchReset], None] | None — The fetch data stream was reset; see FetchReset.
• on_goaway: Callable[[Goaway], None] | None — GOAWAY on this request stream: re-issue the request (possibly elsewhere).

FetchObject

class FetchObject

One object read from a fetch data stream, with its Location made absolute.

Source

Attributes

• kind: Literal['object']
• group_id: int
• object_id: int
• subgroup_id: int | None — Absent for objects with Forwarding Preference = Datagram.
• priority: int
• properties: list[KeyValuePair]
• payload: bytes

FetchOk

class FetchOk

FETCH_OK (§9.12): the publisher accepted a fetch; its objects follow on a fetch data stream.

Source

Attributes

• kind: Literal['fetch-ok']
• end_of_track: bool — The track is complete and end_location is its final Location.
• end_location: Location — Where the fetched range ends, inclusive (§9.12).
• parameters: MessageParameters
• track_properties: list[KeyValuePair]

FetchReset

class FetchReset

The fetch data stream was reset; delivered entries may be incomplete.

Source

Attributes

• error: BaseException

Goaway

class Goaway

GOAWAY (§9.2): the peer intends to close the session soon and asks for no new requests.

Source

Attributes

• kind: Literal['goaway']
• new_session_uri: str — Where to reconnect; empty means the current URI. A client always sends it empty.
• timeout: int — Milliseconds until the peer closes the session; 0 when it gave none.

IncomingRequestUpdate

class IncomingRequestUpdate

REQUEST_UPDATE on an inbound request stream.

request_id is the request being updated (the stream’s SUBSCRIBE or SUBSCRIBE_NAMESPACE); update_request_id is the update’s own, freshly consumed Request ID (§6.4.2.1).

Source

Attributes

• request_id: int
• update_request_id: int
• parameters: MessageParameters

KeyValuePair

class KeyValuePair

A generic MOQT Key-Value-Pair.

Even types carry a varint value, odd types a byte field. Used for Setup Options and Track/Object Properties (message parameters use per-type encodings instead; see MessageParameters).

Source

Attributes

• type: int
• value: int | bytes

Location

class Location

A particular Object in a Group within a Track (§8.2).

Source

Attributes

• group: int
• object: int

MessageParameters

class MessageParameters

Decoded message parameters.

Unlike Setup Options, message parameters use per-type encodings (uint8 / varint / length-prefixed), so the codec is registry-driven. Per §9.20 an unknown parameter type is a PROTOCOL_VIOLATION (parameters cannot be skipped; the block is bounded by count, not length).

fill_parameters (§9.20.15) asks the publisher to deliver a fill range on a fetch stream alongside the subscription; its presence alone requests the fill. It is a nested parameter scope: only the fill-scoped subset (Table 7) is legal inside, and omitted entries inherit the subscription’s own values.

Source

Attributes

• object_delivery_timeout: int | None

• subgroup_delivery_timeout: int | None

• rendezvous_timeout: int | None

• fill_timeout: int | None

• authorization_tokens: list[bytes] | None — Serialized Token structures (§9.20.2). May repeat.

• expires: int | None

• largest_object: Location | None

• forward: int | None — 0 or 1.

• subscriber_priority: int | None

• location_filter: LocationFilter | None

• group_order: GroupOrder | None

• fill_parameters: MessageParameters | None

• new_group_request: int | None

• range_filters: list[RangeFilter] | None

• track_namespace_prefix: TrackNamespace | None

• include_properties: int | None — INCLUDE_PROPERTIES (§9.20.21): 0 asks the responder to send empty Track Properties. Default 1.

• max_age: int | None — MAX_AGE (0x915C6, the MoQ Timestamp extension; not a draft-22 parameter).

  The subscriber’s max age in milliseconds, 0 to 2^53-1. A publisher that honors it skips, for this subscription only, a group once a newer group supersedes it and either that successor arrived at the publisher’s hop at least this long ago or the group’s media time trails the live edge by at least this much; the newest group never expires. moq-dev main turns it into the subscription’s max_delay. A session sends it on SUBSCRIBE and REQUEST_UPDATE only when the peer’s SETUP declared SETUP_OPTION.MAX_AGE, and never on FETCH, TRACK_STATUS or SUBSCRIBE_NAMESPACE; see MoqtSession.max_age_supported.

MoqtObject

class MoqtObject

An object delivered on a subscription’s subgroup stream.

Source

Attributes

• group_id: int
• object_id: int
• subgroup_id: int
• priority: int | None — Publisher priority; None inherits the subscription’s priority.
• status: ObjectStatus
• properties: list[KeyValuePair]
• payload: bytes

MoqtPublishSession

class MoqtPublishSession

The publish-side session driver. Create one with create_moqt_publish_session.

Source

Attributes

• ready: asyncio.Future[None] — Resolves when the server’s SETUP has been received; fails if the session ends first.
• namespace_solicited: bool | None (property) — Whether a current solicitation covers our namespace; None before any is accepted.

MoqtPublishSession.close

def close(self, close_code: int = SESSION_ERROR.NO_ERROR, reason: str = '') -> None

Orderly stop.

End every live subscription (PUBLISH_DONE TRACK_ENDED, then FIN) and retract the announce (NAMESPACE_DONE), give those writes a bounded window to reach the wire (closing a WebTransport discards queued data), then give the peer a bounded window to close each ended subscription’s request stream, which it does once it has consumed the PUBLISH_DONE, and close the transport. Returns immediately; the transport closes within CLOSE_FLUSH_TIMEOUT plus CLOSE_END_ACK_TIMEOUT, and sooner against a peer that answers promptly.

Neither bound waits for acknowledgement, so the writes can still be in flight when the transport closes. A caller that has ended its tracks and must not close while their data streams or ends are unacknowledged waits first, for as long as it chooses, on wait_ends_acknowledged and on its data streams; the relay publisher does, for up to its largest write deadline.

MoqtPublishSession.destroy

def destroy(self, error: BaseException | None = None) -> None

Drain outgoing streams, preserving an optional local closure error.

MoqtPublishSession.announce

def announce(self, track_namespace: Sequence[str]) -> None

Announce the namespace on every matching solicitation, current and future.

The wire write additionally waits for the first register_track: an announce invites immediate SUBSCRIBEs, and an empty registry would answer them with a terminal DOES_NOT_EXIST. Result via on_announced / on_announce_ended.

MoqtPublishSession.register_track

def register_track(self, options: RegisterTrackOptions) -> RegisteredTrack

Register one servable track; the peer’s SUBSCRIBEs are answered from this registry.

MoqtPublishSession.wait_ends_acknowledged

async def wait_ends_acknowledged(self) -> None

Return once the peer acknowledged every subscription end queued so far: its PUBLISH_DONE and FIN.

An end that failed, or whose stream was reset, counts as done. On a transport that does not report acknowledgement, an end counts once its FIN was written. Never raises; the caller bounds the wait.

MoqtPublishSession.open_uni_stream

async def open_uni_stream(self, track_alias: int) -> SendStream

Open a unidirectional data stream for the live subscription bound to track_alias.

The stream counts toward that subscription’s PUBLISH_DONE Stream Count, and a stream still open when the subscription ends is reset before PUBLISH_DONE is written (§9.9). Raises RuntimeError once the session is closing, when no live subscription holds the alias, and when the subscription ends while the stream is opening (the stream is reset first).

MoqtPublishSessionCallbacks

class MoqtPublishSessionCallbacks

Synchronous callbacks a publish session calls as its state changes. Every one is optional.

Source

Attributes

• on_ready: Callable[[list[KeyValuePair]], None] | None — The server’s SETUP arrived; the session is fully established.

• on_goaway: Callable[[Goaway], None] | None — GOAWAY (control stream or a request stream): stop initiating, migrate.

• on_announced: Callable[[TrackNamespace], None] | None — The announced namespace reached the wire on an accepted SUBSCRIBE_NAMESPACE solicitation.

  The peer can now route SUBSCRIBEs to it. NAMESPACE entries have no acknowledgement of their own; this is the strongest “announced” fact the transport draft offers.

• on_announce_ended: Callable[[AnnounceEnded], None] | None — The solicitation stream carrying the announce ended while the session was still live.

  The peer rejected the announce (unauthorized suffix) or withdrew its namespace interest. Either way the ingest path is gone.

• on_subscribe: Callable[[Subscribe], None] | None — The peer subscribed to a registered track (already answered SUBSCRIBE_OK).

• on_subscribe_end: Callable[[int], None] | None — A subscription ended (the peer’s FIN or reset, or our PUBLISH_DONE); the argument is its request id.

• on_subscribe_failed: Callable[[SubscribeFailed], None] | None — A SUBSCRIBE to a registered track could not be answered; see SubscribeFailed.

• on_track_binding: Callable[[TrackBinding], None] | None — A registered track’s effective alias binding changed; see TrackBinding.

• on_request_update: Callable[[IncomingRequestUpdate], None] | None — The peer sent REQUEST_UPDATE on one of its requests; see IncomingRequestUpdate.

• on_closed: Callable[[Closed], None] | None — The session ended: transport closed, or a fatal protocol error.

MoqtPublishSessionConfig

class MoqtPublishSessionConfig

How create_moqt_publish_session sets up a publish session.

Source

Attributes

• setup_options: Sequence[KeyValuePair] — Extra Setup Options to send (an implementation identifier is added automatically).
• implementation_name: str — Identifies this implementation in SETUP (§9.1.5).
• callbacks: MoqtPublishSessionCallbacks

MoqtSession

class MoqtSession

The subscribe-side session driver. Create one with create_moqt_session.

Source

Attributes

• ready: asyncio.Future[None] — Resolves when the server’s SETUP has been received; fails if the session ends first.

• max_age_supported: bool (property) — Whether the server’s SETUP declared subscriber max age (the MoQ Timestamp extension’s MAX_AGE option).

  When it did, parameters.max_age goes out on SUBSCRIBE and REQUEST_UPDATE. While it is false, and before the server’s SETUP has arrived, the session leaves max_age off them, as the extension requires; it never sends it on FETCH, TRACK_STATUS or SUBSCRIBE_NAMESPACE. moq-dev main declares it; moq-relay 0.17.x does not, and on draft-17 and later would fail a SUBSCRIBE that carried the parameter.

MoqtSession.close

def close(self, close_code: int = SESSION_ERROR.NO_ERROR, reason: str = '') -> None

Close the transport session. Sends no GOAWAY (spec/behavior.md, “Relay interop deviations”).

MoqtSession.destroy

def destroy(self, error: BaseException | None = None) -> None

End locally; an error also fails requests awaiting their first response.

MoqtSession.subscribe

def subscribe(
    self,
    options: MoqtSubscribeOptions,
    handlers: SubscriptionHandlers | None = None,
) -> Subscription

Send SUBSCRIBE (§9.6) on a new request stream and return its handle at once.

Results arrive through handlers, which must be synchronous (TypeError otherwise).

MoqtSession.fetch

def fetch(
    self,
    options: MoqtSubscribeOptions,
    handlers: FetchHandlers | None = None,
) -> FetchHandle

Send FETCH (§9.11) on a new request stream and return its handle at once.

The range is options.parameters.location_filter, read with fetch rules; results and entries arrive through handlers.

MoqtSession.track_status

def track_status(
    self,
    options: MoqtSubscribeOptions,
    handlers: TrackStatusHandlers | None = None,
) -> None

Send TRACK_STATUS (§9.13) on a new request stream; the answer arrives through handlers.

MoqtSession.subscribe_namespace

def subscribe_namespace(
    self,
    options: MoqtSubscribeNamespaceOptions,
    handlers: NamespaceSubscriptionHandlers | None = None,
) -> NamespaceSubscription

Send SUBSCRIBE_NAMESPACE (§9.15) on a new request stream and return its handle at once.

The peer answers REQUEST_OK, then names the namespaces under the prefix with NAMESPACE / NAMESPACE_DONE on the same stream. A relay that pushes unsolicited PUBLISH_NAMESPACE instead only stops doing so when SETUP carried SETUP_OPTION.SOLICIT.

MoqtSessionCallbacks

class MoqtSessionCallbacks

Synchronous callbacks a subscribe-side session calls as its state changes. Every one is optional.

Source

Attributes

• on_ready: Callable[[list[KeyValuePair]], None] | None — The server’s SETUP arrived; the session is fully established.

• on_goaway: Callable[[Goaway], None] | None — GOAWAY on the control stream: stop initiating requests, migrate.

• on_incoming_publish: Callable[[Publish, PublishResponder], None] | None — A server-initiated PUBLISH arrived. Call exactly one of the responder’s methods.

  Absent, the session rejects with UNINTERESTED (a subscribe-only client).

• on_closed: Callable[[Closed], None] | None — The session ended: transport closed, or a fatal protocol error.

MoqtSessionConfig

class MoqtSessionConfig

How create_moqt_session sets up a subscribe-side session.

Source

Attributes

• setup_options: Sequence[KeyValuePair] — Extra Setup Options to send (an implementation identifier is added automatically).

• implementation_name: str — Identifies this implementation in SETUP (§9.1.5).

• unknown_alias_timeout: float — Seconds to hold an incoming data stream whose Track Alias has no registered subscription yet.

  Objects can outrace SUBSCRIBE_OK; §11.3.1 allows brief buffering, and expired streams are dropped.

• request_timeout: float — Seconds to wait for a request’s first response before failing it; 0 disables the bound.

  The draft expects implementations to bound control exchanges (§12.2’s CONTROL_MESSAGE_TIMEOUT); without this a relay that accepts the stream and then goes quiet leaves the request pending forever.

• callbacks: MoqtSessionCallbacks

MoqtSubscribeNamespaceOptions

class MoqtSubscribeNamespaceOptions

The prefix and parameters for a SUBSCRIBE_NAMESPACE (§9.15).

Source

Attributes

• track_namespace_prefix: Sequence[str] — Namespace prefix to discover; NAMESPACE entries name namespaces relative to it.
• parameters: MessageParameters | None — None is the same as an empty MessageParameters.

MoqtSubscribeOptions

class MoqtSubscribeOptions

The track a SUBSCRIBE, FETCH or TRACK_STATUS names, and the request’s parameters.

MoqtFetchOptions is the same class: for FETCH, parameters.location_filter is the range, read with fetch rules (§3.3.1).

Source

Attributes

• track_namespace: Sequence[str]
• track_name: str
• parameters: MessageParameters | None — None is the same as an empty MessageParameters.

NamespaceSubscription

class NamespaceSubscription

A live SUBSCRIBE_NAMESPACE request returned by MoqtSession.subscribe_namespace.

Source

Attributes

• request_id: int (property) — The Request ID the SUBSCRIBE_NAMESPACE carried (§6.4.2.1).

NamespaceSubscription.cancel

def cancel(self, reason: object = None) -> None

Tear the request down; the peer withdraws nothing it already sent. Idempotent.

NamespaceSubscriptionHandlers

class NamespaceSubscriptionHandlers

Callbacks for MoqtSession.subscribe_namespace. Every one is optional.

NAMESPACE and NAMESPACE_DONE entries name a namespace by its suffix relative to the requested prefix; an empty suffix is the prefix itself.

Source

Attributes

• on_ok: Callable[[], None] | None — REQUEST_OK arrived: the peer accepted the request, and entries follow on the same stream.

• on_error: Callable[[RequestError], None] | None — The request failed before or instead of REQUEST_OK: REQUEST_ERROR, or a stream failure or timeout.

• on_namespace: Callable[[list[str]], None] | None — NAMESPACE (§9.16): a namespace under the prefix is available.

  A peer may name one again (moq-relay does when its route changes); each call is passed on.

• on_namespace_done: Callable[[list[str]], None] | None — NAMESPACE_DONE (§9.17): a namespace the peer named is no longer available.

  Also called for each namespace still active when the accepted request ends (§9.15), just before on_end. One for a namespace never named closes the session with PROTOCOL_VIOLATION (§9.15).

• on_end: Callable[[BaseException | None], None] | None — The accepted request ended; no entry follows.

  None when the peer finished its side of the stream, the transport’s error when the stream failed. Not called for a local cancel, after REQUEST_ERROR, or when the session ends.

• on_goaway: Callable[[Goaway], None] | None — GOAWAY on this request stream: re-issue the request (possibly elsewhere).

NextObjectFilter

class NextObjectFilter

{Largest.Group, Largest.Object + 1}, the live edge, which can begin mid-group.

Type 0x05 on draft-22, the two-field spelling {0,0} on draft-20.

Source

Attributes

• type: Literal['next-object']

NoLocationFilter

class NoLocationFilter

No location constraint, and the same as sending no LOCATION_FILTER at all.

Type 0x00 on draft-22, zero fields on draft-20. A fetch covers {0,0} up to Largest Object. A subscription passes all objects published or received via upstream subscriptions (§3.3.1); the draft does not say whether that reaches back into objects a relay cached before the subscription, so what an ongoing track delivers is the relay’s choice, and a relay may start at the live edge. RelativeGroupFilter and AbsoluteLocationFilter ask for earlier groups explicitly. In REQUEST_UPDATE it clears the filter.

Source

Attributes

• type: Literal['none']

PublishDone

class PublishDone

PUBLISH_DONE (§9.9): the publisher finished a subscription.

status_code is one of the PUBLISH_DONE status codes; stream_count is how many data streams it opened for the subscription, so a subscriber can tell when all of them have arrived.

Source

Attributes

• kind: Literal['publish-done']
• status_code: int
• stream_count: int | None — None when the publisher could not count its streams (the 2^64-1 sentinel, §9.9).
• reason: str

PublishResponder

class PublishResponder

Answer a server-initiated PUBLISH. Call exactly one of the two methods; later calls are ignored.

Source

PublishResponder.accept

def accept(self) -> None

Send a bare PUBLISH_OK.

Since draft-20 the subscription parameters (FORWARD, LOCATION_FILTER, …) are legal only on PUBLISH and REQUEST_UPDATE, and a parameter in the wrong message is a PROTOCOL_VIOLATION (§9.20.1).

PublishResponder.reject

def reject(
    self,
    error_code: int = REQUEST_ERROR_CODE.UNINTERESTED,
    reason: str = '',
) -> None

Send REQUEST_ERROR with error_code (UNINTERESTED by default) and reason.

RegisterTrackOptions

class RegisterTrackOptions

The track MoqtPublishSession.register_track makes servable.

Source

Attributes

• track_namespace: Sequence[str]
• track_name: str
• track_properties: Sequence[KeyValuePair] — Track metadata is explicit; raw tracks have no implied timestamp unit.

RegisteredTrack

class RegisteredTrack

One servable track.

end() ends every live subscription (its data streams closed, then PUBLISH_DONE TRACK_ENDED and a FIN on its request stream) and refuses future SUBSCRIBEs and TRACK_STATUS requests with DOES_NOT_EXIST. Idempotent.

Source

Attributes

• track_name

RegisteredTrack.end

def end(self) -> None

End every live subscription on the track and refuse later SUBSCRIBEs. Idempotent.

RegisteredTrack.object_published

def object_published(self, location: Location) -> None

Record that the object at location was handed to the transport.

The session keeps the largest location recorded (group first, then object) and answers TRACK_STATUS with it as LARGEST_OBJECT (draft-22 §9.20.17); before the first record the answer has none. The session does not see the data streams’ contents, so whoever writes them reports here. SUBSCRIBE_OK does not carry it.

RelativeGroupFilter

class RelativeGroupFilter

Start at {Largest.Group + 1 - groups_before_next, 0}, open-ended.

0 is the next group, 1 the start of the current group (a decodable join), N N-1 groups before it. Type 0x01 on draft-22, one field on draft-20.

Source

Attributes

• groups_before_next: int
• type: Literal['relative-group']

RequestError

class RequestError

REQUEST_ERROR (§9.4): the peer refused a request.

The session also reports a request that fails on its stream (FIN before any response, a reset, or the response timeout) in this shape, so a consumer has one failure path.

Source

Attributes

• kind: Literal['request-error']
• error_code: int
• retry_interval: int
• reason: str
• redirect: Redirect | None

SubgroupEnd

class SubgroupEnd

A subgroup stream finished cleanly (FIN).

Source

Attributes

• group_id: int
• subgroup_id: int
• end_of_group: bool

SubgroupReset

class SubgroupReset

A subgroup stream was reset; objects from it may be missing.

Reported for every reset the session receives. A relay that drops a reset instead of forwarding it leaves nothing to report.

Source

Attributes

• group_id: int
• error: BaseException

SubgroupWriter

class SubgroupWriter

Writer for one subgroup stream. Create one with create_subgroup_writer.

Source

Attributes

• group_id

SubgroupWriter.write_object

async def write_object(
    self,
    *,
    object_id: int,
    payload: bytes,
    properties: Sequence[KeyValuePair] | None = None,
    status: ObjectStatus = 'normal',
) -> None

Serialize one object. Returns when the transport accepted the bytes.

object_id is absolute and must increase monotonically within the stream. status is only meaningful for zero-length payloads (§11.1.1).

SubgroupWriter.fin

async def fin(self) -> None

FIN the stream: the group is complete.

The FIN counts as final only once the transport has taken it. Under real backpressure it can stay queued for a long time, and the stream has to remain abortable for as long as it does: otherwise a group whose operation was cancelled meanwhile is still delivered complete, and the caller waiting here is never released.

SubgroupWriter.wait_acknowledged

async def wait_acknowledged(self) -> None

After fin: return once the peer acknowledged the whole stream, FIN included.

On a transport that does not report acknowledgement this returns at once. Raises when the stream was reset or stopped, or the session ended, before the acknowledgement.

SubgroupWriter.abort

def abort(self, reason: object = None) -> None

Reset the stream; delivered objects may be incomplete (drop path).

A FIN the transport has not taken yet is reset too, and the pending fin raises reason: the send half turns a closed-but-untransmitted stream into a RESET_STREAM. So is a FIN that was taken but not yet acknowledged, which is how a settling group’s SUBGROUP_DELIVERY_TIMEOUT resets it (§5.2); wait_acknowledged then raises. Once the whole stream is acknowledged the transport has nothing left to reset. Idempotent.

SubscribeFailed

class SubscribeFailed

What on_subscribe_failed receives: the session could not build its SUBSCRIBE_OK.

The peer was answered REQUEST_ERROR INTERNAL_ERROR instead. The track stays registered and the session stays up, but a cause that lies in the track (its properties) fails every subscription to it. error is the encoder’s failure.

Source

Attributes

• request_id: int
• track_name: str
• error: BaseException

SubscribeOk

class SubscribeOk

SUBSCRIBE_OK (§9.7): the publisher accepted a subscription.

track_alias is the alias its data streams carry; parameters and track_properties are as the publisher sent them.

Source

Attributes

• kind: Literal['subscribe-ok']
• track_alias: int
• parameters: MessageParameters
• track_properties: list[KeyValuePair]

Subscription

class Subscription

A live subscription request.

MOQT has no UNSUBSCRIBE message: teardown is the request stream’s lifecycle (§6.4.2.3).

Source

Attributes

• request_id: int (property) — The Request ID the SUBSCRIBE carried (§6.4.2.1).

Subscription.update

def update(self, parameters: MessageParameters) -> None

Modify the subscription (REQUEST_UPDATE, §9.5).

Subscription.cancel

def cancel(self, reason: object = None) -> None

Tear the subscription down.

SubscriptionHandlers

class SubscriptionHandlers

Callbacks for one subscription. Every one is optional and must be synchronous.

An async callback raises TypeError at subscribe time; a callback that returns an awaitable has it discarded with a RuntimeWarning. Relay’s subscribe_events() is the async way in.

Source

Attributes

• on_ok: Callable[[SubscribeOk], None] | None — The publisher accepted the subscription.

• on_error: Callable[[RequestError], None] | None — The request failed: a REQUEST_ERROR, or a stream failure or timeout in the same shape.

• on_object: Callable[[MoqtObject], None] | None — An object arrived on one of the subscription’s data streams.

• on_subgroup_end: Callable[[SubgroupEnd], None] | None — A subgroup stream finished cleanly.

• on_subgroup_reset: Callable[[SubgroupReset], None] | None — A subgroup stream was reset; see SubgroupReset.

• on_done: Callable[[PublishDone], None] | None — PUBLISH_DONE arrived: the publisher finished the subscription.

• on_update_ok: Callable[[], None] | None — Response to a REQUEST_UPDATE sent on this subscription.

• on_state_notify: Callable[[MessageParameters], None] | None — PUBLISH_STATE_NOTIFY (§9.10): the publisher changed subscription state on its own.

  The parameters carry only what changed, with LARGEST_OBJECT marking where. Informational; the session sends no reply.

• on_goaway: Callable[[Goaway], None] | None — GOAWAY on this request stream: re-issue the request (possibly elsewhere).

TrackBinding

class TrackBinding

A track’s effective alias binding changed.

The most recent live subscription wins the binding (a replaced one ends with PUBLISH_DONE TRACK_ENDED, a clean end for that request, covering the peer’s resubscribe races), and only while its Forward State permits data (forward=0 subscribes without authorizing it; REQUEST_UPDATE toggles it). None means no live forwarding subscription, so no data streams may be opened for the track.

The two delivery timeouts are that subscription’s values in milliseconds (draft-22 §5.2, §9.20.3, §9.20.4): SUBSCRIBE’s parameters as updated by every REQUEST_UPDATE since, None where the subscriber never sent one, and always None without a binding. A REQUEST_UPDATE that changes either is reported as a binding with the same alias.

Source

Attributes

• track_name: str
• track_alias: int | None
• object_delivery_timeout: int | None
• subgroup_delivery_timeout: int | None

TrackStatusHandlers

class TrackStatusHandlers

Callbacks for one TRACK_STATUS request. Both are optional.

Source

Attributes

• on_ok: Callable[[TrackStatusOk], None] | None — The publisher answered with the track’s status.
• on_error: Callable[[RequestError], None] | None — The request failed: a REQUEST_ERROR, or a stream failure or timeout in the same shape.

TrackStatusOk

class TrackStatusOk

The answer to TRACK_STATUS (a REQUEST_OK): the track’s current parameters and properties.

Source

Attributes

• parameters: MessageParameters
• track_properties: list[KeyValuePair]

Protocols

MoqTrackSource

class MoqTrackSource(Protocol)

The subscribe / fetch / track-status subset of a session, plus closure observation, without shutdown.

Source

Attributes

• closed: asyncio.Future[Closed] (property) — Resolves when the owning connection ends; an error denotes failure.

MoqTrackSource.subscribe

def subscribe(
    self,
    options: MoqtSubscribeOptions,
    handlers: SubscriptionHandlers | None = None,
) -> Subscription

Subscribe to a track, as MoqtSession.subscribe does; the caller owns the request.

MoqTrackSource.fetch

def fetch(
    self,
    options: MoqtSubscribeOptions,
    handlers: FetchHandlers | None = None,
) -> FetchHandle

Fetch a range of a track, as MoqtSession.fetch does; the caller owns the request.

MoqTrackSource.track_status

def track_status(
    self,
    options: MoqtSubscribeOptions,
    handlers: TrackStatusHandlers | None = None,
) -> None

Ask for a track’s status, as MoqtSession.track_status does.

MoqtTransport

class MoqtTransport(Protocol)

Structural subset of a WebTransport session the session drivers need.

Source

MoqtTransport.incoming_unidirectional_streams

def incoming_unidirectional_streams(self) -> AsyncIterator[ReceiveStream]

Iterate peer-initiated unidirectional streams until the transport ends.

MoqtTransport.incoming_bidirectional_streams

def incoming_bidirectional_streams(self) -> AsyncIterator[BidirectionalStream]

Iterate peer-initiated bidirectional streams until the transport ends.

MoqtTransport.create_unidirectional_stream

async def create_unidirectional_stream(self) -> SendStream

Open a unidirectional stream to the peer.

MoqtTransport.create_bidirectional_stream

async def create_bidirectional_stream(self) -> BidirectionalStream

Open a bidirectional stream to the peer.

MoqtTransport.close

def close(self, close_code: int = 0, reason: str = '') -> None

Close the session. Idempotent; a transport that already ended ignores it.

MoqtTransport.wait_closed

async def wait_closed(self) -> None

Return once the session ended cleanly (either side); raise the error that ended it otherwise.

ReceiveStream

class ReceiveStream(Protocol)

The receiving half of a stream.

Source

ReceiveStream.read

async def read(self) -> bytes | None

Return the next non-empty chunk, or None at a clean end of stream. Raises on a reset.

ReceiveStream.cancel

def cancel(self, reason: object = None) -> None

Stop reading (STOP_SENDING). Later read calls end; the peer sees the stream stopped.

SendStream

class SendStream(Protocol)

The sending half of a stream.

Source

SendStream.write

async def write(self, data: bytes) -> None

Queue data; returns once the transport accepted it.

Raises once the stream is closed, aborted, or stopped by the peer.

SendStream.close

async def close(self) -> None

FIN the stream after every accepted write.

SendStream.abort

def abort(self, reason: object = None) -> None

Reset the stream (RESET_STREAM); queued bytes may never arrive.

Enums

PUBLISH_DONE_STATUS

class PUBLISH_DONE_STATUS(IntEnum)

PUBLISH_DONE status codes (§12.4).

Source

Attributes

• INTERNAL_ERROR
• UNAUTHORIZED
• TRACK_ENDED
• GOING_AWAY
• TOO_FAR_BEHIND
• EXPIRED
• UPDATE_FAILED
• EXCESSIVE_LOAD
• MALFORMED_TRACK

REQUEST_ERROR_CODE

class REQUEST_ERROR_CODE(IntEnum)

REQUEST_ERROR codes (§16.11.2).

Source

Attributes

• INTERNAL_ERROR
• UNAUTHORIZED
• TIMEOUT
• NOT_SUPPORTED
• MALFORMED_AUTH_TOKEN
• EXPIRED_AUTH_TOKEN
• GOING_AWAY
• EXCESSIVE_LOAD
• DOES_NOT_EXIST
• INVALID_RANGE
• MALFORMED_TRACK
• UNINTERESTED
• PREFIX_OVERLAP
• NAMESPACE_TOO_LARGE
• UNSUPPORTED_EXTENSION
• REDIRECT
• CONFLICTING_FILTERS
• INVALID_FILTER

SESSION_ERROR

class SESSION_ERROR(IntEnum)

Session-termination codes (§12.2).

Source

Attributes

• NO_ERROR
• INTERNAL_ERROR
• UNAUTHORIZED
• PROTOCOL_VIOLATION
• INVALID_REQUEST_ID
• DUPLICATE_TRACK_ALIAS
• KEY_VALUE_FORMATTING_ERROR
• INVALID_PATH
• MALFORMED_PATH
• GOAWAY_TIMEOUT
• CONTROL_MESSAGE_TIMEOUT
• DATA_STREAM_TIMEOUT
• AUTH_TOKEN_CACHE_OVERFLOW
• DUPLICATE_AUTH_TOKEN_ALIAS
• MALFORMED_AUTH_TOKEN
• UNKNOWN_AUTH_TOKEN_ALIAS
• EXPIRED_AUTH_TOKEN
• INVALID_AUTHORITY
• MALFORMED_AUTHORITY
• TOO_MANY_REQUEST_UPDATES

Exceptions

MoqtProtocolError

class MoqtProtocolError(Exception)

A wire-protocol violation detected while encoding or decoding MOQT data.

code is the session-termination code (§12.2) the session driver should close the transport with; it defaults to PROTOCOL_VIOLATION.

Source

Attributes

• code

Types

FetchStreamEntry

FetchStreamEntry: TypeAlias = FetchObject | FetchEndOfRange

Source

GroupOrder

GroupOrder: TypeAlias = Literal['ascending', 'descending']

Source

LocationFilter

LocationFilter: TypeAlias = ...

Which Objects a subscription or fetch covers.

Draft-22 opens the LOCATION_FILTER value with a Location Filter Type that selects the fields after it; draft-20 length-prefixes up to four varints and the number present selects the meaning.

Source

MoqtFetchOptions

MoqtFetchOptions: TypeAlias = MoqtSubscribeOptions

FETCH (§9.11) takes the SUBSCRIBE shape.

The range is parameters.location_filter, read with fetch rules (§3.3.1): a relative start counts back from Largest Object, an omitted end stops there, no filter means {0,0} up to Largest Object. A subscription reads the same filter with subscription rules; see NoLocationFilter.

Source

MoqtProtocolId

MoqtProtocolId: TypeAlias = Literal['moqt-22', 'moqt-20']

A protocol identifier the core implements.

Source

ObjectStatus

ObjectStatus: TypeAlias = Literal['normal', 'end-of-group', 'end-of-track']

Source

TrackNamespace

TrackNamespace: TypeAlias = list[str]

Track namespace as an ordered tuple of UTF-8 fields. Byte strings on the wire.

Source

Constants

MAX_NAMESPACE_FIELDS

MAX_NAMESPACE_FIELDS = 32

Source

MAX_OBJECT_PAYLOAD_LENGTH

MAX_OBJECT_PAYLOAD_LENGTH = 16 * 1024 * 1024

Ceiling on a single object’s payload.

Unlike control messages, bounded by their 16-bit frame length, a data stream’s payload length is an unbounded varint that feeds straight into StreamReader.read_bytes, which buffers until satisfied. A malformed or hostile relay declaring 2^53 would exhaust memory before the read ever completed, so the declaration is rejected before any allocation. 16 MiB clears any real LOC frame.

Source

MAX_TRACK_PROPERTIES_LENGTH

MAX_TRACK_PROPERTIES_LENGTH = MAX_CONTROL_MESSAGE_LENGTH - 9 - 1

The largest encoded track property block every SUBSCRIBE_OK this session sends can carry.

The properties ride in SUBSCRIBE_OK, a control message whose length field is 16 bits (§9). The rest of its body is the track alias, a varint of up to 9 bytes (it is the peer’s request ID, so it is unknown until a subscription arrives), and the count of message parameters, of which this session sends none (1 byte). A longer block registers but can never answer a subscription.

Source

MOQT_PROTOCOL_ID

MOQT_PROTOCOL_ID: MoqtProtocolId = 'moqt-22'

The preferred protocol identifier, and the codec’s default when a caller names none.

Source

MOQT_PROTOCOL_IDS

MOQT_PROTOCOL_IDS: tuple[MoqtProtocolId, ...] = ('moqt-22', 'moqt-20')

ALPN / WebTransport protocol identifiers the core implements, in the order a client offers them.

§6.2: moqt- plus the draft number. Version negotiation happens at connection time (WT-Available-Protocols in WebTransport, ALPN in native QUIC), not in SETUP. Draft-22 is the default; draft-20 stays as a fallback for a peer that does not speak draft-22. The two differ on the wire only in the LOCATION_FILTER parameter (§9.20.9), so one codec serves both, and every function that encodes or decodes message parameters takes the negotiated identifier.

Source