shiguredo_moqt

draft-ietf-moq-transport-22 に基づく Sans I/O / no_std な Media over QUIC Transport (MOQT) ライブラリ。

特徴

• Sans I/O: I/O を完全に分離。Tokio / async-std / 同期 I/O など任意の環境で使用可能
• no_std 対応: alloc のみ必要 (std 非依存)
• 依存は 4 つのみ: hashbrown / noflate / nojson / base64ct (暗号処理は optional feature aws-lc-rs)
• Session 状態機械: 1 本の Transport Session に閉じた endpoint-local な状態管理
• インクリメンタルデコード: フレーム境界に依存せず受信バッファに蓄積しながらデコード
• 音声の再生補助: 時間伸縮、目標遅延の学習、鳴らす時刻の決定、A/V 同期の遅延制御 (no_std / Sans I/O)
• 対応仕様: MOQT draft-22 / LOC draft-04 / MSF draft-01 / C4M draft-01

バージョン情報

• crate 名: shiguredo_moqt
• バージョン: 2026.0.0
• Rust Edition: 2024
• 最小 Rust バージョン: 1.93
• ライセンス: Apache-2.0

重要: このライブラリは re-export を行わない。型は定義元モジュールから直接 import すること。

// OK
use shiguredo_moqt::session::core::Session;
use shiguredo_moqt::session::types::{SessionEvent, Transport};
use shiguredo_moqt::stream::subgroup::SubgroupHeader;
use shiguredo_moqt::message::common::TrackNamespace;

// NG (re-export は存在しない)
use shiguredo_moqt::session::Session;
use shiguredo_moqt::stream::SubgroupHeader;

モジュール構成

モジュール	役割
message	制御メッセージ。message::common に TrackNamespace / Location
message_parameter	Message Parameters
parameter	SETUP Options
decoder	MessageDecoder (制御メッセージのインクリメンタルデコード)
stream	data stream / datagram のヘッダとデコーダ / エンコーダ
session	session::core::Session 状態機械
loc	LOC Properties
object_properties	Object-scoped Properties
track_properties	Track Properties
msf	MSF Catalog / Timeline / URI
c4m	C4M (CAT / CWT) のトークン発行 / 検証と moqt クレームの認可
c4m::cbor / c4m::cose / c4m::crypto	CBOR / COSE のコーデックと暗号 trait
c4m::cat / c4m::jwk / c4m::jwt / c4m::dpop	CAT / JWK / JWT / DPoP proof
varint	可変長整数 (vi64)
name	Namespace / Track Name の文字列表現
grease	GREASE 値の生成・判定
subgroup_tracker	Subgroup 再オープン禁止の検証
playout	時間伸縮、目標遅延の学習、鳴らす時刻の決定、A/V 同期の遅延制御、共通の時間軸
error	エラー型とエラーコード定数

クイックスタート

制御メッセージの encode / decode

use shiguredo_moqt::error::MessageError;
use shiguredo_moqt::message::{ControlMessage, Setup};
use shiguredo_moqt::parameter::SetupOptions;

fn main() -> Result<(), MessageError> {
    let msg = ControlMessage::Setup(Setup {
        options: SetupOptions::new(),
    });
    let bytes = msg.encode()?;

    let (decoded, consumed) = ControlMessage::decode(&bytes)?;
    let _ = (decoded, consumed);
    Ok(())
}

セッション

use shiguredo_moqt::parameter::SetupOptions;
use shiguredo_moqt::session::core::Session;
use shiguredo_moqt::session::types::{SessionError, SessionEvent, Transport};

fn main() -> Result<(), SessionError> {
    let mut session = Session::new_client(Transport::Quic, SetupOptions::new())?;
    while let Some(event) = session.poll_event() {
        match event {
            SessionEvent::SendControl(msg) => {
                // 制御ストリームへ msg.encode()? を書き出す
                let _ = msg;
            }
            SessionEvent::Established => {
                // 両側の SETUP が完了
            }
            SessionEvent::CloseSession(e) => {
                // QUIC CONNECTION_CLOSE 等で接続を閉じる
                let _ = e;
            }
            _ => {}
        }
    }
    Ok(())
}

制御メッセージ (message)

fn encode(&self) -> Result<Vec<u8>, MessageError>
fn decode(buf: &[u8]) -> Result<(ControlMessage, usize), MessageError>

Type (varint) | Length (u16 big-endian) | Message Body 形式。decode は (メッセージ, 消費バイト数) を返す。

主な variant:

variant	用途
Setup(Setup)	セッション確立
Goaway(Goaway)	セッション終了通知
RequestOk / RequestError	共通応答 (PUBLISH への PUBLISH_OK は RequestOk の別名)
Subscribe / SubscribeOk	SUBSCRIBE
Publish / PublishDone	PUBLISH
PublishStateNotify	状態通知
RequestUpdate	更新要求
Fetch / FetchOk	FETCH
TrackStatus	TRACK_STATUS (送信側と受信側の両方)
Unsupported	定義済みだが本ライブラリが扱わない型 (本体は生バイト列のまま保持)

補助型:

use shiguredo_moqt::message::common::{Location, TrackNamespace};

// TrackNamespace: フィールド数 0〜32、空フィールド禁止、合計 4096 バイト以下
fn new(fields: Vec<Vec<u8>>) -> Result<TrackNamespace, MessageError>
fn fields(&self) -> &[Vec<u8>]
fn is_session_level(&self) -> bool
fn is_single_period(&self) -> bool
fn byte_length(&self) -> usize

// Location: group_id / object_id の公開フィールドを持つ。Copy / Ord
pub struct Location {
    pub group_id: u64,
    pub object_id: u64,
}

// ReasonPhrase: 1024 バイト超でエラー
fn new(s: impl Into<String>) -> Result<ReasonPhrase, MessageError>
fn as_str(&self) -> &str

インクリメンタルデコード (decoder)

use shiguredo_moqt::decoder::MessageDecoder;

let mut decoder = MessageDecoder::new();
decoder.push(&data);
if let Some(msg) = decoder.try_decode_message()? {
    // msg を Session::recv_control 等に渡す
}

fn new() -> MessageDecoder
fn push(&mut self, data: &[u8])
fn try_decode_message(&mut self) -> Result<Option<ControlMessage>, MessageError>
fn try_decode_varint(&mut self) -> Result<Option<u64>, MessageError>

• Ok(Some) = デコード成功・バッファから消費済み
• Ok(None) = データ不足 (追加の push が必要)
• Err = デコードエラー

try_decode_varint は受信 uni stream の先頭 stream type の読み取りに使う。

セッション (session)

生成と基本操作

use shiguredo_moqt::parameter::SetupOptions;
use shiguredo_moqt::session::core::Session;
use shiguredo_moqt::session::types::Transport;

fn new_client(transport: Transport, options: SetupOptions) -> Result<Session, SessionError>
fn new_server(transport: Transport, options: SetupOptions) -> Result<Session, SessionError>
fn poll_event(&mut self) -> Option<SessionEvent>
fn tick(&mut self, now_ms: u64)
fn close(&mut self, code: u64, reason: &'static str)
fn state(&self) -> SessionState
fn role(&self) -> Role
fn transport(&self) -> Transport

tick(now_ms) には単調増加ミリ秒時刻を渡す。control message / GOAWAY のセッション全体の タイムアウトと data stream timeout を評価する。control message / GOAWAY と FETCH stream の data stream timeout の満了時は CloseSession をキューへ積む。subgroup stream の data stream timeout ではセッションを閉じず、当該 stream に StopSendingDataStream { stream_id, error_code } を積む (draft-ietf-moq-transport-22 §12.2 (Session Termination Codes) の "it MAY send a STOP_SENDING on that stream or terminate the subscription" に従い、購読は生かしたまま timeout した stream だけを止める。 §6.6 (Termination) は単一購読の条件をセッションエラーへ昇格させることを MAY で認めており、 クローズも仕様違反ではないが、他の購読への影響を避けるため STOP_SENDING を選ぶ)。 timeout 後に同じ stream へ届く Object / FIN / RESET_STREAM は peer_alias_retention_ms の 保持期間の間 no-op で吸収する。 あわせて subscription の delivery timeout は ResetDataStream { error_code: STREAM_DELIVERY_TIMEOUT } を積み、 alias tombstone / 破棄対象 stream id の保持期間は期限切れの片付けだけを行う。 request stream 上の GOAWAY の timeout はセッションを閉じず、 当該 request の ResetRequestStream { error_code: STREAM_GOING_AWAY } を積む。さらに期限到達時には、 まだ終端していない request を終端して RequestTerminated { reason: GoawayTimeout } を 1 回だけ積む (reset 後は PUBLISH_DONE を送れないため)。既に終端済みの request には reset イベントだけを積む。

タイマーと Auth Token:

fn control_message_timeout_ms(&self) -> Option<u64>
fn set_control_message_timeout_ms(&mut self, timeout_ms: Option<u64>)
fn data_stream_timeout_ms(&self) -> Option<u64>
fn set_data_stream_timeout_ms(&mut self, timeout_ms: Option<u64>)
fn peer_auth_token_cache(&self) -> &AuthTokenCache
fn peer_max_auth_token_cache_size(&self) -> u64
fn peer_alias_retention_ms(&self) -> u64
fn set_peer_alias_retention_ms(&mut self, retention_ms: u64)

利用者が match / 引数で使う enum

pub enum Transport { Quic, WebTransport, Qmux }
pub enum Role { Client, Server }
pub enum SessionState { LocalSetupSent, Established, Closing, Closed }
pub enum TrackRole { Publisher, Subscriber }
pub enum SubscriptionInitiator { Subscriber, Publisher }
pub enum SubscriptionState { Pending, Established, Terminated }
pub enum RequestKind { Subscribe, Publish, Fetch, TrackStatus }
pub enum RequestStreamEnd {
    Fin,
    Reset { error_code: u64, reliable_size: Option<u64> },
}
pub enum FetchState { Pending, Established, Terminated }
pub enum TrackDataAcceptance { Accepted, UnknownTrackAlias, Discarded, FilteredOut }
pub enum DatagramAcceptance { Object(TrackDataAcceptance), Padding }

DataStreamResetReason は fn error_code(self) -> u64 で対応する Stream Reset Error Code を取得できる。

SessionEvent

poll_event() が返すイベント。I/O 層はこれを受けて実際の送信・切断を行う。

variant	意味
SendControl(ControlMessage)	制御ストリームに書く
SendRequest { request_id, message }	新規 bidi request stream を開いて書く
SendOnStream { request_id, message, fin }	既存 bidi request stream に書く (fin で FIN)
Established	両側 SETUP 完了
CloseSession(SessionError)	接続を閉じる
RequestUpdateReceived { request_id, parameters }	REQUEST_UPDATE 受信
PublishStateNotifyReceived { request_id, parameters }	PUBLISH_STATE_NOTIFY 受信
RequestOkReceived { request_id, request_kind, parameters }	REQUEST_OK 受信
FetchOkReceived { request_id, end_location, end_of_track }	FETCH_OK 受信
PublishDoneReceived { request_id, status_code, stream_count, reason }	PUBLISH_DONE 受信
RequestErrorReceived { request_id, error_code, retry_interval, reason, redirect }	REQUEST_ERROR 受信
RequestTerminated { request_id, kind, reason }	request が Terminated に遷移
GoawayReceived { new_session_uri, timeout, on_request_stream }	GOAWAY 受信
SendPaddingStream { length }	指定長のパディングストリームを送る
SendPaddingDatagram { length }	指定長のパディングデータグラムを送る
ResetDataStream { stream_id, error_code, reliable_size }	data stream を reset する
StopSendingDataStream { stream_id, error_code }	受信 data stream に STOP_SENDING を送る (subgroup stream の DATA_STREAM_TIMEOUT。error_code は STREAM_CANCELLED。timeout 後の遅延到着は保持期間の間 no-op)
OpenFillFetchStream { request_id, subscription_request_id, start, end }	fill fetch stream を開く (request_id は FETCH_HEADER に載せる起因メッセージの Request ID、subscription_request_id は fill 対象購読、start / end は解決済みの fill range)
ResetRequestStream { request_id, error_code }	request stream の送信方向を reset する
StopSendingRequestStream { request_id, error_code }	request stream の受信方向に STOP_SENDING を送る
FinishRequestStream { request_id }	request stream の送信方向を FIN で閉じる

bidi request stream は方向ごとに独立して閉じる (draft-ietf-moq-transport-22 §6.4.2.2 (Graceful Request Stream Closure))。 RequestTerminated { reason: PeerStreamFin } は peer の FIN の到着と同時に発行されるとは限らず、 自側が SUBSCRIBE / FETCH / TRACK_STATUS の responder、または PUBLISH を送った側 (publisher 役) で 最終メッセージを未送信の場合は peer の FIN と自側が最終メッセージとともに送る FIN の両方が揃った時点で発行される。 request stream 上の GOAWAY の timeout 満了で自側が終端した場合は RequestTerminated { reason: GoawayTimeout } が発行される (draft-ietf-moq-transport-22 §9.2 (GOAWAY) の "SHOULD reset the stream with GOING_AWAY" を送信方向のみで満たすため、PUBLISH_DONE は送れず保留分は破棄される)。 reset するのは送信方向だけなので peer の送信方向は開いたままであり、終端済み request へ届く 1 回目の peer FIN / RESET_STREAM は no-op で吸収される。peer が reset を観測する前に開いた FETCH 応答の遅延 data stream は破棄対象 stream として吸収し、 fill fetch stream は subscription への帰属を保ったまま受理する (破棄後の終端通知は no-op)。peer が GOAWAY / reset を観測する前に 送った応答 (SUBSCRIBE_OK / REQUEST_OK / REQUEST_ERROR / FETCH_OK / PUBLISH_DONE) と REQUEST_UPDATE / PUBLISH_STATE_NOTIFY は、各ハンドラがそのメッセージに該当する仕様の検証 (送信者・種別・自側の役割・パラメータスコープ・FORWARD・Track Alias・Request ID・ MAX_REQUEST_UPDATES・auth token・End Location) を行ったうえで、状態遷移せず受理する (subscription の遅延 REQUEST_ERROR は呼び出し元の §9.4.1 Redirect 検証のみで受理する)。遅延した REQUEST_ERROR では RequestErrorReceived のみを通知する (他の応答は状態遷移もイベント発行もしない)。ローカル終端済みの REQUEST_UPDATE では Range Filter の内容検証を省略する (draft §3.3.2 の MUST は REQUEST_ERROR による拒否であり、 送信方向を reset 済みで応答を送れないため満たせない)。終端で PUBLISH_OK / REQUEST_UPDATE_OK の 区別が state から失われるため、REQUEST_OK のパラメータは両 context の許可集合で検証する (自側 publisher で initiator の場合、PUBLISH_OK context でも LARGEST_OBJECT を許容する残差がある)。state テーブルから破棄済みの request (forget_* 後) へのメッセージは request 種別と状態が失われているため受理せず、 従来どおりセッションを閉じる (遅延メッセージの到着が止まってから forget_* を呼ぶこと。peer の close 通知は no-op で吸収されるため通知されない)。 メッセージ内容が仕様の MUST に反する場合や、それ以外のメッセージ (PUBLISH 等) は従来どおり セッションを閉じる。

受信 API

// 制御ストリーム / bidi request stream
fn recv_control(&mut self, msg: ControlMessage) -> Result<(), SessionError>
fn recv_control_stream_type(&mut self, stream_type: u64) -> Result<(), SessionError>
fn recv_control_stream_closed(&mut self, end: RequestStreamEnd) -> Result<(), SessionError>
fn recv_request(&mut self, msg: ControlMessage) -> Result<(), RecvRequestError>
fn recv_request_stream_closed(&mut self, request_id: u64, end: RequestStreamEnd) -> Result<(), SessionError>
fn recv_stream_message(&mut self, request_id: u64, msg: ControlMessage) -> Result<(), SessionError>

// data stream / datagram
fn recv_data_stream_type(
    &mut self,
    stream_id: DataStreamId,
    stream_type: u64,
) -> Result<DataStreamType, RecvDataStreamError>
fn recv_data_stream_stop_sending(
    &mut self,
    stream_id: DataStreamId,
    error_code: Option<u64>,
) -> Result<(), SessionError>
fn recv_subgroup_header(
    &mut self,
    stream_id: DataStreamId,
    header: &SubgroupHeader,
) -> Result<TrackDataAcceptance, SessionError>
fn recv_subgroup_object(
    &mut self,
    stream_id: DataStreamId,
    object: &DecodedSubgroupObject,
) -> Result<TrackDataAcceptance, SessionError>
fn recv_fetch_header(&mut self, stream_id: DataStreamId, header: &FetchHeader) -> Result<(), SessionError>
fn recv_fetch_entry(&mut self, stream_id: DataStreamId) -> Result<(), SessionError>
fn recv_fetch_data_stream_closed(&mut self, request_id: u64, end: RequestStreamEnd) -> Result<(), SessionError>
fn fetch_stop_sending_received(&mut self, request_id: u64, error_code: Option<u64>) -> Result<(), SessionError>
fn recv_datagram(&mut self, raw: &[u8]) -> Result<DatagramAcceptance, SessionError>
fn recv_object_datagram(&mut self, datagram: &ObjectDatagram) -> Result<TrackDataAcceptance, SessionError>
fn recv_padding_datagram(&mut self) -> Result<(), SessionError>
fn recv_data_stream_closed(&mut self, stream_id: DataStreamId, end: RequestStreamEnd) -> Result<(), SessionError>
fn report_mid_object_fin(&mut self, stream_id: DataStreamId) -> Result<(), SessionError>

DataStreamId(pub u64) は受信 uni data stream を呼び出し側が管理するための識別子。

recv_subgroup_object() は Object 単位のフィルタ再適用結果を TrackDataAcceptance で返す。 FilteredOut / Discarded の Object は Application へ渡さないが、wire 上の payload は デコーダから読み出して消費する必要がある。

recv_request_stream_closed(rid, RequestStreamEnd::Fin) は、自側が SUBSCRIBE / FETCH / TRACK_STATUS の responder、または PUBLISH を送った側 (publisher 役) で Established のとき request を終端しない。この場合 RequestTerminated は自側が最終メッセージを fin: true で送った時点 (PUBLISH_DONE、単独の REQUEST_ERROR) で発行される。 request stream 上の GOAWAY の timeout が満了した場合も request を終端し、 RequestTerminated { reason: GoawayTimeout } が発行される (詳細は RequestTerminated の項を参照)。 このとき publisher 役 fetch の open 中の outgoing FETCH data stream は Session が reset しないため、 アプリが send_fetch_data_stream_closed で閉じること。 RequestStreamEnd::Reset は cancel として即時に終端する (draft-ietf-moq-transport-22 §6.4.2.3 (Request Cancellation and Rejection))。 error_code は Option<u64> であり、None は「アプリケーションエラーコード無し」を表す (draft-ietf-webtrans-http3-16 §4.4 (Resetting Data Streams)。WebTransport over HTTP/3 経路で HTTP/3 のコードを MOQT のコードへ戻せない場合に None になる)。QUIC 直接接続と QMux は QUIC のコード space を共有し、WebTransport over HTTP/2 は capsule がコードをそのまま運ぶため、 いずれも常に Some。 recv_data_stream_stop_sending(stream_id, error_code) と fetch_stop_sending_received(request_id, error_code) の error_code も同じ意味である。 受信したコードは Session::stopped_outgoing_subgroup_error_code(request_id, track_alias, group_id, subgroup_id) -> Option<Option<u64>> (外側が再オープン禁止の有無、内側がコードの有無) と Fetch::peer_stop_sending_error_code (外側がデータストリーム側の STOP_SENDING 受理の有無、 内側がコードの有無) から参照できる。

送信 API

// subscription
fn send_subscribe(
    &mut self,
    track_namespace: TrackNamespace,
    track_name: Vec<u8>,
    parameters: MessageParameters,
) -> Result<u64, SendRequestError>
fn send_publish(
    &mut self,
    track_namespace: TrackNamespace,
    track_name: Vec<u8>,
    track_alias: u64,
    parameters: MessageParameters,
    track_properties: TrackProperties,
) -> Result<u64, SendRequestError>
fn send_subscribe_ok(
    &mut self,
    request_id: u64,
    track_alias: u64,
    parameters: MessageParameters,
    track_properties: TrackProperties,
) -> Result<(), SessionError>
fn send_request_update(&mut self, request_id: u64, parameters: MessageParameters) -> Result<(), SessionError>
fn send_publish_state_notify(&mut self, request_id: u64, parameters: MessageParameters) -> Result<(), SessionError>
fn send_publish_done(
    &mut self,
    request_id: u64,
    status_code: u64,
    stream_count: u64,
    reason: ReasonPhrase,
) -> Result<(), SessionError>

// FETCH
fn send_fetch(
    &mut self,
    track_namespace: TrackNamespace,
    track_name: Vec<u8>,
    parameters: MessageParameters,
) -> Result<u64, SendRequestError>
fn send_fetch_ok(
    &mut self,
    request_id: u64,
    end_of_track: u8,
    end_location: Location,
    parameters: MessageParameters,
    track_properties: TrackProperties,
) -> Result<(), SessionError>

// TRACK_STATUS
fn send_track_status(
    &mut self,
    track_namespace: TrackNamespace,
    track_name: Vec<u8>,
    parameters: MessageParameters,
) -> Result<u64, SendRequestError>

// 受信した TRACK_STATUS への応答は send_request_ok (TRACK_STATUS_OK) と
// send_request_error (REQUEST_ERROR) で送る。どちらも送信後に bidi stream を FIN で閉じる。

// 共通応答
fn send_request_ok(
    &mut self,
    request_id: u64,
    parameters: MessageParameters,
    track_properties: TrackProperties,
) -> Result<(), SessionError>
fn send_request_error(
    &mut self,
    request_id: u64,
    error_code: u64,
    retry_interval: u64,
    reason: ReasonPhrase,
    redirect: Option<Redirect>,
) -> Result<(), SessionError>

data plane 送信 API

fn send_subgroup_header(
    &mut self,
    stream_id: DataStreamId,
    request_id: u64,
    header: &SubgroupHeader,
) -> Result<(), SessionError>
fn send_subgroup_object(
    &mut self,
    stream_id: DataStreamId,
    object_id: u64,
    properties_bytes: Option<&[u8]>,
) -> Result<(), SendRequestError>
fn send_data_stream_closed(&mut self, stream_id: DataStreamId, end: RequestStreamEnd) -> Result<(), SessionError>
fn send_fetch_header(&mut self, stream_id: DataStreamId, request_id: u64) -> Result<(), SessionError>
fn send_fill_fetch_header(&mut self, stream_id: DataStreamId, fill_request_id: u64) -> Result<(), SessionError>
fn send_fetch_object(&mut self, stream_id: DataStreamId) -> Result<(), SessionError>
fn send_fetch_data_stream_closed(&mut self, stream_id: DataStreamId) -> Result<(), SessionError>
fn send_object_datagram(
    &mut self,
    request_id: u64,
    group_id: u64,
    object_id: u64,
    properties_data: Option<Vec<u8>>,
    status: Option<u64>,
    end_of_group: bool,
) -> Result<(), SendRequestError>
fn send_data_stream_stop_sending(&mut self, stream_id: DataStreamId) -> Result<(), SessionError>
fn reset_outgoing_data_stream(
    &mut self,
    stream_id: DataStreamId,
    reason: DataStreamResetReason,
) -> Result<(), SessionError>

forget API

fn forget_subscription(&mut self, request_id: u64) -> Option<Subscription>
fn forget_fetch(&mut self, request_id: u64) -> Option<Fetch>
fn forget_track_status(&mut self, request_id: u64) -> Option<TrackStatusEntry>

Session::forget_subscription は、保留中の PUBLISH_DONE (Subscription::pending_publish_done) が ある間は呼ばないこと。呼ぶと PUBLISH_DONE が送信されず、draft-ietf-moq-transport-22 §9.5.1 (Updating Subscriptions) の MUST を果たせない。順序契約と受信側の追跡状態の扱いの全文は Session::forget_subscription の doc を参照すること。前提条件は API ごとに異なる (forget_subscription は cleanup_ready、forget_fetch / forget_track_status はそれぞれの終端 条件) ため、呼ぶ前に各 API の doc を確認すること。

公開定数

定数	値	説明
PUBLISH_DONE_STREAM_COUNT_UNKNOWN	u64::MAX	Stream Count が不明な場合の sentinel
DEFAULT_PEER_ALIAS_RETENTION_MS	5000	キャンセル済み peer publisher alias の tombstone 保持期間 (ms)
PUBLISHER_PRIORITY_DEFAULT	128	DEFAULT_PUBLISHER_PRIORITY の既定値
DEFAULT_PUBLISHER_GROUP_ORDER_ASCENDING	0x1	DEFAULT_PUBLISHER_GROUP_ORDER の既定値
MAX_NEW_SESSION_URI_LENGTH	8192	GOAWAY の New Session URI 最大長

エラー型

pub struct SessionError {
    pub code: u64,
    pub reason: &'static str,
}

pub enum SendRequestError {
    PeerGoawayReceived,
    LocalFilterMismatch,
    LocalDatagramTimeout,
    Session(SessionError),
}

pub enum RecvRequestError {
    BeforeSessionEstablished,
    Session(SessionError),
}

pub enum RecvDataStreamError {
    BeforeSessionEstablished,
    InvalidInput(SessionError),
    Session(SessionError),
}

• SendRequestError は as_session_error() で SessionError を取り出せる (PeerGoawayReceived / LocalFilterMismatch / LocalDatagramTimeout は None)。これらは wire コードを持たないため公開 API に渡す必要はない (渡しても各レジストリの *_INTERNAL_ERROR に置換される)
• RecvRequestError / RecvDataStreamError は BeforeSessionEstablished なら session state に影響しない

data stream / datagram (stream)

受信ストリーム種別

use shiguredo_moqt::stream::DataStreamType;

pub enum DataStreamType { Fetch, Subgroup, Padding }

fn classify_data_stream_type(type_id: u64) -> Option<DataStreamType>

ヘッダ型

use shiguredo_moqt::stream::subgroup::{SubgroupHeader, SubgroupIdMode, SubgroupObject};
use shiguredo_moqt::stream::fetch::{
    FetchHeader, FetchPriorContext, FetchSubgroupIdMode, FetchStreamEntry, FetchStreamObject,
};
use shiguredo_moqt::stream::datagram::ObjectDatagram;

pub enum SubgroupIdMode { Zero, FirstObjectId, Explicit(u64) }
pub enum FetchSubgroupIdMode { Zero, PreviousSame, PreviousPlusOne, Explicit(u64) }
pub enum FetchPriorContext { First, NoPriorActualObject, HasPriorObject }

型ごとのシグネチャは次のとおり (一般化しない。実際の src/stream/ と一致させること)。

// ヘッダ (draft-ietf-moq-transport-22 §11.3.1 / §11.4.1)
fn SubgroupHeader::encode(&self) -> Vec<u8>
fn SubgroupHeader::decode(buf: &[u8]) -> Result<(SubgroupHeader, usize), MessageError>
fn FetchHeader::encode(&self) -> Vec<u8>
fn FetchHeader::decode(buf: &[u8]) -> Result<(FetchHeader, usize), MessageError>

// Object (Properties の有無は引数で渡す)
fn SubgroupObject::encode(
    &self,
    has_properties: bool,
    properties_data: Option<&[u8]>,
    buf: &mut Vec<u8>,
) -> Result<(), MessageError>
fn SubgroupObject::decode(
    buf: &[u8],
    has_properties: bool,
) -> Result<(SubgroupObject, Option<Vec<u8>>, usize), MessageError>

fn FetchStreamEntry::encode(
    &self,
    properties_data: Option<&[u8]>,
    prior_context: FetchPriorContext,
    buf: &mut Vec<u8>,
) -> Result<(), MessageError>
fn FetchStreamEntry::decode(
    buf: &[u8],
    prior_context: FetchPriorContext,
) -> Result<(FetchStreamEntry, usize), MessageError>

fn FetchStreamObject::encode(
    &self,
    properties_data: Option<&[u8]>,
    prior_context: FetchPriorContext,
    buf: &mut Vec<u8>,
) -> Result<(), MessageError>
// FetchStreamObject に公開の decode は無い。`FetchStreamEntry::decode` が
// flags を読んでから内部の decode_after_flags を通す

// Datagram (1 datagram = 1 バッファ)
fn ObjectDatagram::encode(&self) -> Result<Vec<u8>, MessageError>
fn ObjectDatagram::decode(buf: &[u8]) -> Result<(ObjectDatagram, usize), MessageError>

ObjectDatagram::properties_data と send_object_datagram の properties_data は、Properties Length varint を含む生バイト列である (SubgroupObject / FetchStreamObject と同じ規約)。Datagram では Properties Length = 0 はプロトコル違反であり、encode / 送信 API が拒否する。

send_object_datagram の end_of_group は Object Datagram の END_OF_GROUP bit に対応し、その位置より大きい Object ID の Object が同じ Group に存在しないことを宣言する。status を指定した datagram と END_OF_GROUP bit は同時に指定できず、encode / 送信 API が拒否する。

デコーダ / エンコーダ

use shiguredo_moqt::stream::decoder::{SubgroupStreamDecoder, FetchStreamDecoder};
use shiguredo_moqt::stream::encoder::{FetchStreamEncoder, FetchObjectInput};

// SubgroupStreamDecoder
fn new() -> SubgroupStreamDecoder
fn push(&mut self, data: &[u8])
fn try_decode_header(&mut self) -> Result<Option<SubgroupHeader>, MessageError>
fn try_decode_object(&mut self) -> Result<Option<DecodedSubgroupObject>, MessageError>
fn try_read_payload(&mut self) -> Option<Vec<u8>>
fn consume_payload(&mut self, length: u64) -> Result<(), MessageError>
fn resolved_subgroup_id(&self) -> Option<u64>
fn finish(&self) -> Result<(), MessageError>

// FetchStreamDecoder
fn new() -> FetchStreamDecoder
fn new_with_group_order(group_order: u8) -> Result<FetchStreamDecoder, MessageError>
fn set_subgroup_final_object(
    &mut self,
    group_id: u64,
    subgroup_id: u64,
    final_object_id: u64,
) -> Result<(), MessageError>
fn push(&mut self, data: &[u8])
fn try_decode_header(&mut self) -> Result<Option<FetchHeader>, MessageError>
fn try_decode_entry(&mut self) -> Result<Option<DecodedFetchEntry>, MessageError>
fn try_read_payload(&mut self) -> Option<Vec<u8>>
fn consume_payload(&mut self, length: u64) -> Result<(), MessageError>
fn finish(&self) -> Result<(), MessageError>

// FetchStreamEncoder
fn new(request_id: u64) -> FetchStreamEncoder
fn new_with_group_order(request_id: u64, group_order: u8) -> Result<FetchStreamEncoder, MessageError>
fn encode_header(&self) -> Vec<u8>
fn encode_object(
    &mut self,
    input: &FetchObjectInput,
    properties_data: Option<&[u8]>,
    buf: &mut Vec<u8>,
) -> Result<(), MessageError>
fn encode_end_of_non_existent_range(&mut self, group_id: u64, object_id: u64, buf: &mut Vec<u8>) -> Result<(), MessageError>
fn encode_end_of_unknown_range(&mut self, group_id: u64, object_id: u64, buf: &mut Vec<u8>) -> Result<(), MessageError>
fn encode_end_of_timed_out_range(&mut self, group_id: u64, object_id: u64, buf: &mut Vec<u8>) -> Result<(), MessageError>

• SubgroupStreamDecoder は try_decode_object / consume_payload、FetchStreamDecoder は try_decode_entry / consume_payload とメソッド名が異なる
• ペイロード本体はデコーダの責務外。payload_length バイト分を読み出したら consume_payload を呼ぶ
• Object Properties は DecodedSubgroupObject::properties_bytes と DecodedFetchObject::properties_bytes の どちらも Properties Length varint 込みの生バイト列であり、LocProperties::decode にそのまま渡せる。 wire に現れた varint をそのまま保持するため、Properties Length = 0 は最小形なら 0x00 の 1 バイトになる
• Properties フィールドの有無は Subgroup では SUBGROUP_HEADER の Type Flags の PROPERTIES ビット、 FETCH では Serialization Flags (draft-ietf-moq-transport-22 §11.4.1.1 Table 10) の bit 0x20 が示す。 どちらの経路も、無い場合は None、ある場合は Some になる
• FetchStreamEncoder は前回オブジェクトとの差分 (デルタ圧縮) を内部状態で判断する
• FetchStreamEncoder::new は Group Order を Ascending (0x01) に固定する。new_with_group_order は 0x01 (Ascending) と 0x02 (Descending) を受け付け、それ以外は ProtocolViolation を返す (FetchStreamDecoder::new_with_group_order と対称)
• FetchStreamEncoder はペイロードを出力しない。返したバイト列の後ろに呼び出し側がペイロードを結合する

SETUP Options (parameter)

use shiguredo_moqt::parameter::{SetupOption, SetupOptionValue, SetupOptions};

pub struct SetupOptions(/* Vec<SetupOption> */);

fn new() -> SetupOptions
fn push(&mut self, p: SetupOption)
fn len(&self) -> usize
fn is_empty(&self) -> bool
fn encode(&self, buf: &mut Vec<u8>) -> Result<(), MessageError>
fn decode(buf: &[u8]) -> Result<(SetupOptions, usize), MessageError>

// アクセサ
fn path(&self) -> Option<&[u8]>
fn authorization_tokens(&self) -> Vec<&AuthorizationToken>
fn max_auth_token_cache_size(&self) -> Option<u64>
fn authority(&self) -> Option<&[u8]>
fn max_request_updates(&self) -> Option<u64>
fn max_filter_ranges(&self) -> Option<u64>

pub struct SetupOption {
    pub option_type: u64,
    pub value: SetupOptionValue,
}

pub enum SetupOptionValue {
    VarInt(u64),
    Bytes(Vec<u8>),
    AuthorizationToken(AuthorizationToken),
}

Setup Option 型定数:

定数	値
SETUP_OPTION_PATH	0x01
SETUP_OPTION_AUTHORIZATION_TOKEN	0x03
SETUP_OPTION_MAX_AUTH_TOKEN_CACHE_SIZE	0x04
SETUP_OPTION_AUTHORITY	0x05
SETUP_OPTION_MAX_FILTER_RANGES	0x06
SETUP_OPTION_MOQT_IMPLEMENTATION	0x07
SETUP_OPTION_MAX_REQUEST_UPDATES	0x08

Message Parameters (message_parameter)

use shiguredo_moqt::message_parameter::{
    AuthorizationToken, LocationFilter, LocationFilterUpdate, MessageParameter, MessageParameters,
    MessageParameterValue,
};

pub struct MessageParameter {
    pub param_type: u64,
    pub value: MessageParameterValue,
}

pub enum MessageParameterValue {
    Uint8(u8),
    VarInt(u64),
    Location { group: u64, object: u64 },
    LengthPrefixed(Vec<u8>),
    LocationFilter(LocationFilter),
    FillParameters(MessageParameters),
    AuthorizationToken(AuthorizationToken),
    TrackNamespacePrefix(TrackNamespace),
}

pub struct MessageParameters(/* Vec<MessageParameter> */);

fn new() -> MessageParameters
fn push(&mut self, p: MessageParameter)
fn len(&self) -> usize
fn is_empty(&self) -> bool
fn as_slice(&self) -> &[MessageParameter]
fn encode(&self, buf: &mut Vec<u8>) -> Result<(), MessageError>
fn decode(buf: &[u8]) -> Result<(MessageParameters, usize), MessageError>

主なアクセサ:

fn object_delivery_timeout(&self) -> Option<u64>            // 0x02
fn subgroup_delivery_timeout(&self) -> Option<u64>          // 0x06
fn expires(&self) -> Option<u64>                            // 0x08
fn largest_object(&self) -> Option<(u64, u64)>              // 0x09
fn fill_timeout(&self) -> Option<u64>                       // 0x0A
fn forward(&self) -> Option<u8>                             // 0x10
fn subscriber_priority(&self) -> Option<u8>                 // 0x20
fn location_filter_typed(&self) -> Result<Option<LocationFilter>, MessageError>  // 0x21
fn location_filter_update(&self) -> Result<LocationFilterUpdate, MessageError>  // 0x21
fn group_order(&self) -> Option<u8>                         // 0x22
fn new_group_request(&self) -> Option<u64>                  // 0x32
fn track_namespace_prefix(&self) -> Option<&TrackNamespace> // 0x34
fn include_properties(&self) -> Option<u8>                  // 0x35

主なパラメータ型定数:

定数	値
PARAM_OBJECT_DELIVERY_TIMEOUT	0x02
PARAM_AUTHORIZATION_TOKEN	0x03
PARAM_RENDEZVOUS_TIMEOUT	0x04
PARAM_SUBGROUP_DELIVERY_TIMEOUT	0x06
PARAM_EXPIRES	0x08
PARAM_LARGEST_OBJECT	0x09
PARAM_FILL_TIMEOUT	0x0A
PARAM_FORWARD	0x10
PARAM_SUBSCRIBER_PRIORITY	0x20
PARAM_LOCATION_FILTER	0x21
PARAM_GROUP_ORDER	0x22
PARAM_FILL_PARAMETERS	0x23
PARAM_SUBGROUP_FILTER	0x25
PARAM_OBJECTID_FILTER	0x26
PARAM_PRIORITY_FILTER	0x27
PARAM_OBJECT_PROPERTY_FILTER	0x28
PARAM_TRACK_PROPERTY_FILTER	0x29
PARAM_NEW_GROUP_REQUEST	0x32
PARAM_TRACK_NAMESPACE_PREFIX	0x34
PARAM_INCLUDE_PROPERTIES	0x35

AuthorizationToken は Delete / Register / UseAlias / UseValue の variant を持つ。各メソッドは pub(crate) のため、利用側は variant を直接構築して MessageParameterValue に格納する。

LocationFilter は Location Filter Type (vi64) が後続フィールドを定める符号化の typed 表現で、次の variant を持つ。

• NoFilter (Type 0x00)
• RelativeGroup { start_group } (Type 0x01)
• AbsoluteStart { start } (Type 0x02)
• AbsoluteRange { start, end_group_delta } (Type 0x03)
• AbsoluteRangeWithEnd { start, end_group_delta, end_object } (Type 0x04)
• NextObject (Type 0x05)

MessageParameters::location_filter_update() は LocationFilterUpdate の Unchanged (パラメータ省略) / Removed (Type 0x00 = フィルタなし) / Set(filter) を返す。

LOC Properties (loc)

use shiguredo_moqt::loc::{LocProperties, LocProperty, LocPropertyValue};

pub enum LocPropertyValue {
    VarInt(u64),
    Bytes(Vec<u8>),
}

pub struct LocProperty {
    pub prop_id: u64,
    pub value: LocPropertyValue,
}

fn new() -> LocProperties
fn push(&mut self, prop: LocProperty)
fn iter(&self) -> impl Iterator<Item = &LocProperty>
fn encode(&self) -> Result<Vec<u8>, MessageError>
fn decode(buf: &[u8]) -> Result<(LocProperties, usize), MessageError>

// アクセサ
fn timestamp(&self) -> Option<u64>
fn timescale(&self) -> Option<u64>
fn video_frame_marking(&self) -> Option<&[u8]>
fn audio_level(&self) -> Option<u64>
fn video_config(&self) -> Option<&[u8]>
fn audio_config(&self) -> Option<&[u8]>

定数	値
PROP_TIMESTAMP	0x10
PROP_TIMESCALE	0x08
PROP_VIDEO_FRAME_MARKING	0x09
PROP_AUDIO_LEVEL	0x0C
PROP_VIDEO_CONFIG	0x0D
PROP_AUDIO_CONFIG	0x0F

偶数 ID は varint 値、奇数 ID は長さ付きバイト列。

Object / Track Properties

object_properties

use shiguredo_moqt::object_properties::{
    ObjectProperties, ObjectProperty, ObjectPropertyValue, ObjectPropertyTracker,
    ObjectFieldTracker,
};

fn new() -> ObjectProperties
fn push(&mut self, prop: ObjectProperty)
fn encode(&self, buf: &mut Vec<u8>) -> Result<(), MessageError>
fn decode(buf: &[u8]) -> Result<(ObjectProperties, usize), MessageError>

// アクセサ
fn object_delivery_timeout(&self) -> Option<u64>
fn subgroup_delivery_timeout(&self) -> Option<u64>
fn prior_group_id_gap(&self) -> Option<u64>
fn prior_object_id_gap(&self) -> Option<u64>
fn immutable_properties(&self) -> Option<&[u8]>
fn find_varint(&self, prop_type: u64) -> Option<u64>

定数	値
PROP_PRIOR_GROUP_ID_GAP	0x3C
PROP_PRIOR_OBJECT_ID_GAP	0x3E

PROP_OBJECT_DELIVERY_TIMEOUT / PROP_SUBGROUP_DELIVERY_TIMEOUT / PROP_IMMUTABLE_PROPERTIES は track_properties から import する。

track_properties

use shiguredo_moqt::track_properties::{
    TrackProperties, TrackProperty, TrackPropertyValue,
};

fn new() -> TrackProperties
fn push(&mut self, p: TrackProperty)
fn encode(&self, buf: &mut Vec<u8>) -> Result<(), MessageError>
fn decode(buf: &[u8]) -> Result<TrackProperties, MessageError>

// アクセサ
fn dynamic_groups(&self) -> Option<u64>
fn default_publisher_priority(&self) -> Option<u8>
fn default_publisher_group_order(&self) -> Option<u8>
fn object_delivery_timeout(&self) -> Option<u64>
fn subgroup_delivery_timeout(&self) -> Option<u64>
fn has_unknown_mandatory(&self) -> bool

定数	値
PROP_OBJECT_DELIVERY_TIMEOUT	0x02
PROP_MAX_CACHE_DURATION	0x04
PROP_SUBGROUP_DELIVERY_TIMEOUT	0x06
PROP_IMMUTABLE_PROPERTIES	0x0B
PROP_DEFAULT_PUBLISHER_PRIORITY	0x0E
PROP_DEFAULT_PUBLISHER_GROUP_ORDER	0x22
PROP_DYNAMIC_GROUPS	0x30

TrackProperties::decode はバッファ全体を読み、(Self, usize) ではなく Self を返す点に注意。

MSF (msf)

Catalog

use shiguredo_moqt::error::MessageError;
use shiguredo_moqt::msf::MsfCatalogDocument;

fn example(bytes: &[u8]) -> Result<(), MessageError> {
    // Full / Delta を自動判別
    let doc = MsfCatalogDocument::decode(bytes)?;
    let encoded = doc.encode()?;
    let _ = encoded;
    Ok(())
}

型は MsfCatalogDocument { Full(MsfCatalog), Delta(MsfDeltaUpdate) }、要素は MsfCatalog / MsfDeltaOperation / MsfDeltaUpdate / MsfTrack / MsfCloneTrack / MsfRemoveTrack / MsfInitData / MsfBuffers / MsfPackaging、定数は MSF_CATALOG_TRACK_NAME / MSF_VERSION。

pub enum MsfCatalogDocument {
    Full(MsfCatalog),
    Delta(MsfDeltaUpdate),
}

fn apply_delta(
    &mut self,
    delta: &MsfDeltaUpdate,
    catalog_namespace: Option<&str>,
) -> Result<(), MessageError>

MsfTrack は name / namespace / packaging / codec / mime_type / width / height / framerate / bitrate / samplerate / channel_config などを持つ単一構造体。Codec / Video / Audio は draft 上の論理的な分類。

Timeline

use shiguredo_moqt::msf::{
    TimelineEncodingOptions, MsfMediaTimeline, MsfEventTimeline,
    encode_media_timeline, decode_media_timeline,
    encode_event_timeline, decode_event_timeline,
};

pub struct TimelineEncodingOptions { pub gzip: bool }

fn encode_media_timeline(
    timeline: &MsfMediaTimeline,
    options: TimelineEncodingOptions,
) -> Result<Vec<u8>, MessageError>
fn decode_media_timeline(data: &[u8]) -> Result<MsfMediaTimeline, MessageError>
fn encode_event_timeline(
    timeline: &MsfEventTimeline,
    options: TimelineEncodingOptions,
) -> Result<Vec<u8>, MessageError>
fn decode_event_timeline(data: &[u8]) -> Result<MsfEventTimeline, MessageError>

decode_* は先頭の gzip magic (0x1F 0x8B) を検出すると自動展開する。

URI

use shiguredo_moqt::msf::uri::{parse_msf_uri, parse_msf_fragment};

fn parse_msf_uri(uri: &str) -> Result<MsfUri, MessageError>
fn parse_msf_fragment(fragment: &str) -> Result<MsfFragment, MessageError>

MsfFragment は parameter_values / connection_types / wallclock_ranges / mediatime_ranges / location_ranges / c4m_tokens のアクセサを持つ。

C4M (c4m)

draft-ietf-moq-c4m-01 (CAT-4-MOQT) の認可トークンを扱う。CBOR (RFC 8949) / COSE (RFC 9052) / CWT (RFC 8392) / CAT (CTA-5007-B) を含む。

# #[cfg(feature = "aws-lc-rs")]
# fn main() -> Result<(), Box<dyn std::error::Error>> {
use shiguredo_moqt::c4m::cat::{CatToken, CatTokenBuilder};
use shiguredo_moqt::c4m::crypto::aws_lc_rs::AwsLcRsCrypto;
use shiguredo_moqt::c4m::crypto::CoseKey;
use shiguredo_moqt::c4m::{MoqtAction, MoqtClaim, MoqtScope};

let crypto = AwsLcRsCrypto::new();
let key = CoseKey::symmetric([0xab; 32]);

// 発行 (認可サーバー側): compact 形式 / COSE 形式を選べる
let token_text = CatTokenBuilder::new()
    .issuer("https://auth.example.com")
    .audience("https://relay.example.com")
    .expiration(1_700_086_400.0)
    .moqt(
        MoqtClaim::new().scope(
            MoqtScope::new([MoqtAction::Publish, MoqtAction::Fetch])
                .namespace_match(shiguredo_moqt::c4m::NamespaceMatch::Match(
                    shiguredo_moqt::c4m::Match::Exact(b"example.com".to_vec()),
                )),
        ),
    )
    .jwk_thumbprint([0x01; 32])
    .catdpop(300.0, true)
    .build_compact(&crypto, &key)?;

// 検証 (MoQ リレー / endpoint 側)
let token = CatToken::decode(token_text.as_bytes())?;
token.verify(&crypto, &key)?;
token.claims().validate(&shiguredo_moqt::c4m::cat::ClaimValidationOptions {
    reference_time_seconds: 1_700_000_000.0,
    expected_issuers: &["https://auth.example.com"],
    ..Default::default()
})?;
assert!(token
    .claims()
    .authorize(MoqtAction::Publish, &[b"example.com"], b"live"));
# Ok(())
# }

主な API:

// トークンの発行 / 検証
use shiguredo_moqt::c4m::cat::{CatToken, CatTokenBuilder, VerifyOptions};
fn CatTokenBuilder::build_compact(&self, crypto: &C, key: &CoseKey) -> Result<String, CatError>
fn CatTokenBuilder::build_cose(&self, crypto: &C, key: &CoseKey) -> Result<Vec<u8>, CatError>
fn CatToken::decode(input: &[u8]) -> Result<CatToken, CatError>
fn CatToken::verify(&self, crypto: &C, key: &CoseKey) -> Result<(), CatError>
fn CatToken::verify_with(&self, crypto: &C, key: &CoseKey, options: &VerifyOptions) -> Result<(), CatError>
fn CatToken::decode_moqt_auth_token(token_type: u64, value: &[u8]) -> Result<CatToken, CatError>

// moqt クレームの認可
use shiguredo_moqt::c4m::{Match, MoqtAction, MoqtClaim, MoqtScope, NamespaceMatch};
fn MoqtClaim::authorize(&self, action: MoqtAction, namespace: &[&[u8]], track_name: &[u8]) -> bool
fn MoqtScope::allows(&self, action: MoqtAction, namespace: &[&[u8]], track_name: &[u8]) -> bool

// 暗号 trait (feature "aws-lc-rs" で実装を提供する)
use shiguredo_moqt::c4m::crypto::{CoseCrypto, CoseKey, EcCurve, OkpCurve};

// DPoP proof
use shiguredo_moqt::c4m::dpop::{DpopProof, DpopProofBuilder, DpopReplayCache, DpopVerification};
fn DpopProof::decode(input: &str) -> Result<DpopProof, DpopError>
fn DpopProof::verify_against_token(&self, crypto: &C, request: &DpopVerification<'_>, replay_cache: Option<&mut DpopReplayCache>) -> Result<(), DpopError>
fn DpopProof::verify_against_cat_token(&self, crypto: &C, token: &CatToken, action: MoqtAction, namespace: &TrackNamespace, track_name: &[u8], reference_time_seconds: f64, default_window_seconds: f64, replay_cache: Option<&mut DpopReplayCache>) -> Result<(), DpopError>

• 直列化は compact 形式 (draft 付録 A の base64url(protected).base64url(claims).base64url(signature)) と COSE 形式 (CWT タグ + COSE_Sign1 / COSE_Mac0) の両方を扱う
• 暗号実装は aws-lc-rs feature でのみ提供する。既定ビルドは no_std のままで、CoseCrypto trait と鍵表現だけを持つ
• 対応アルゴリズムは HMAC 256/384/512・ES256/384/512・EdDSA (Ed25519)。RSA は未対応

その他

// varint
use shiguredo_moqt::varint;
fn encode(val: u64, buf: &mut Vec<u8>)
fn decode(buf: &[u8]) -> Result<(u64, usize), MessageError>
fn checked_len(len: u64, remaining: usize) -> Result<usize, MessageError>
fn encoded_len(val: u64) -> usize

// name
use shiguredo_moqt::name::{
    parse_name, parse_name_with_percent_encoding, parse_namespace, parse_track_name,
    serialize_name, serialize_namespace, serialize_track_name,
};
fn serialize_name(namespace: &TrackNamespace, track_name: &[u8]) -> String
fn parse_name(s: &str) -> Result<(TrackNamespace, Vec<u8>), NameParseError>
fn parse_name_with_percent_encoding(s: &str) -> Result<(TrackNamespace, Vec<u8>), NameParseError>
fn serialize_namespace(namespace: &TrackNamespace) -> String
fn parse_namespace(s: &str) -> Result<TrackNamespace, NameParseError>
fn serialize_track_name(track_name: &[u8]) -> String
fn parse_track_name(s: &str) -> Result<Vec<u8>, NameParseError>

// grease
use shiguredo_moqt::grease::{generate, is_grease};
fn generate(n: u64) -> Option<u64>
fn is_grease(value: u64) -> bool

// subgroup_tracker
use shiguredo_moqt::subgroup_tracker::{SubgroupStreamState, SubgroupTracker};
fn new() -> SubgroupTracker
fn open(&mut self, track_alias: u64, group_id: u64, subgroup_id: u64) -> Result<(), SessionError>
fn mark_fin(&mut self, track_alias: u64, group_id: u64, subgroup_id: u64, last_object_id: Option<u64>) -> Result<(), SessionError>
fn mark_reset(&mut self, track_alias: u64, group_id: u64, subgroup_id: u64, reliable_size: Option<u64>)
fn mark_stop_sending(&mut self, track_alias: u64, group_id: u64, subgroup_id: u64)
fn get(&self, track_alias: u64, group_id: u64, subgroup_id: u64) -> Option<&SubgroupStreamState>
fn remove_track_alias(&mut self, track_alias: u64)
fn record_priority(
    &mut self,
    track_alias: u64,
    group_id: u64,
    subgroup_id: u64,
    publisher_priority: u8,
) -> Result<(), SessionError>
fn check_object_after_fin(
    &self,
    track_alias: u64,
    group_id: u64,
    subgroup_id: u64,
    object_id: u64,
) -> Option<&'static str>
fn SubgroupStreamState::can_reopen(&self) -> bool

// playout::stretch
use shiguredo_moqt::playout::stretch::{
    compress, conceal, concealment_end_gain, is_time_stretch_available,
};
fn compress(
    channels: &mut [&mut [f32]],
    sample_rate: u32,
    max_length_change_samples: usize,
) -> isize
fn conceal(
    channels: &[&[f32]],
    output: &mut [&mut [f32]],
    sample_rate: u32,
    conceal_us: i64,
) -> isize
fn concealment_end_gain(conceal_us: i64) -> f32
fn is_time_stretch_available(sample_rate: u32) -> bool

// playout::gap
use shiguredo_moqt::playout::gap::{AUDIO_GAP_FADE_US, MaterializedGap, materialize};
fn materialize(
    previous: Option<&[Vec<f32>]>,
    current: &mut [&mut [f32]],
    sample_rate: u32,
    gap_us: i64,
    conceal_us: i64,
) -> MaterializedGap

// playout::buffer
use shiguredo_moqt::playout::buffer::{JITTER_BUFFER_MAX_QUEUED_FRAMES, PlayoutBuffer};
fn new(max_queued_frames: usize, track: Track) -> PlayoutBuffer<T>
fn enqueue(&mut self, item: T, timestamp_us: Option<i64>, timeline: &PlayoutTimeline) -> Vec<T>
fn select(&mut self, now_us: i64, timeline: &PlayoutTimeline) -> PlayoutSelection<T>
fn presentation_time_us(
    &self,
    timestamp_us: Option<i64>,
    timeline: &PlayoutTimeline,
) -> Option<i64>
fn playout_delay_us(&self, timeline: &PlayoutTimeline) -> Option<i64>
fn len(&self) -> usize
fn is_empty(&self) -> bool
fn clear(&mut self) -> Vec<T>

// media_clock
use shiguredo_moqt::media_clock::WallClockMapper;
fn new() -> WallClockMapper
fn observe(&mut self, media_us: i64, wall_clock_us: i64)
fn to_wall_clock_us(&mut self, media_us: i64, fallback_wall_clock_us: Option<i64>) -> Option<i64>
fn offset_us(&self) -> Option<i64>
fn reset(&mut self)

// playout::delay
use shiguredo_moqt::playout::delay::AudioDelayManager;
fn new() -> AudioDelayManager
fn observe(&mut self, arrival_us: i64, capture_us: i64)
fn target_delay_ms(&self) -> i64
fn reset(&mut self)

// playout::scheduler
use shiguredo_moqt::playout::scheduler::{
    AudioPlayoutDecision, AudioPlayoutInput, AudioPlayoutScheduler,
};
fn new() -> AudioPlayoutScheduler
fn schedule(&mut self, input: AudioPlayoutInput) -> AudioPlayoutDecision
fn confirm_stretch(&mut self, applied_us: i64)
fn confirm_concealment(&mut self, applied_us: i64)
fn last_end_us(&self) -> Option<i64>
fn note_underrun(&mut self, now_us: i64)
fn rebases(&self) -> u64
fn drops(&self) -> u64
fn compressed_us(&self) -> i64
fn concealed_us(&self) -> i64
fn concealments(&self) -> u64
fn lateness_us(&self) -> i64
fn reset(&mut self)
fn reset_stats(&mut self)

// playout::timeline
use shiguredo_moqt::playout::timeline::{PlayoutTimeline, TimelineConfig, Track};
fn new() -> PlayoutTimeline
fn with_config(config: TimelineConfig) -> PlayoutTimeline
fn config(&self) -> TimelineConfig
fn observe(&mut self, track: Track, wall_clock_us: i64, timestamp_us: i64)
fn present_us(&self, track: Track, timestamp_us: i64) -> Option<i64>
fn presentation_delay_us(&self, track: Track) -> Option<i64>
fn learned_delay_us(&self, track: Track) -> i64
fn set_target_latency_ms(&mut self, target_latency_ms: i64)
fn target_latency_ms(&self) -> i64
fn limited_us(&self) -> i64
fn record_presentation(&mut self, track: Track, timestamp_us: i64, presented_wall_clock_us: i64)
fn skew_us(&self) -> Option<i64>
fn sharing_bases(&self) -> bool
fn generation(&self) -> u64
fn reset_track(&mut self, track: Track)
fn reset(&mut self)

playout::stretch は 8 kHz / 16 kHz / 32 kHz / 48 kHz に対応する (is_time_stretch_available で確かめられる)。戻り値は長さの変化 (圧縮は負、0 は操作なし)。時間伸長は持たない (この再生は音を伸ばす場面を持たず、足りない区間は conceal が埋める)。 conceal は末尾の 2 周期分の相関から求めた周期で末尾を繰り返し、conceal_us 分を埋めた長さを返す (0 なら output に何も書かない)。利得は末尾へ向けて concealment_end_gain まで下げる。

playout::gap は再生の隙間を材料化する。出力が FIFO (積んだ順に鳴らす) の場合は隙間を積まないと後続が早く鳴って出力が枯れるため、materialize は要求した隙間の全長ぶんの音を必ず返す。conceal_us の分は直前の音の末尾の周期で埋め、残りは無音にする。隙間の末尾と今回の音の先頭は AUDIO_GAP_FADE_US で 0 へ落とし、隙間の先頭は直前の音の最終サンプルから 0 へ向かう重みで繋ぐ。current の先頭にはフェードインが掛かるため &mut で渡す。

playout::delay は到着の遅れの分布の 0.95 分位から目標遅延を決める (まだ観測が無いときは 80 ms)。

playout::scheduler は目標の時刻に従って音を並べ、間に合わない音は時間圧縮で目標へ戻す (AudioPlayoutDecision の Play / Drop を返す)。Play の gap_us は前の音の終わりから今回の開始までの全長 (5 ms や 100 ms で切り捨てない)、conceal_us はそのうち補間してよい長さである。 compress_us を実際に適用したら confirm_stretch へ、隙間を材料化したら confirm_concealment へ実測を返す。出力が枯れたときは、枯れた時間を隙間として二重に数えないよう note_underrun に今の時刻を渡す。並べた音の終わりは last_end_us で読める。

playout::buffer は復号したフレームを表示時刻に合わせて選ぶキューである。表示時刻を過ぎたフレームのうち最新の 1 枚を描き、MAX_PRESENTATION_LAG_US を超えて遅れたフレームは捨てる。表示時刻を決められないフレームと、積んだときの世代が今と違うフレームは届いた順に 1 枚ずつ描く。上限 (JITTER_BUFFER_MAX_QUEUED_FRAMES) を超えたフレームは enqueue が古い方から返す。

media_clock はデバイスが返すメディア時刻を Unix epoch の壁時計へ換算する。observe で「壁時計 − メディア時刻」の最小値を目標にし、to_wall_clock_us でその目標へ向けて 1 回の換算につきメディア時刻の差の半分未満だけ動かす (換算した TIMESTAMP が戻らない)。

playout::timeline は表示時刻 = TIMESTAMP + 基準の遅れ + 表示の遅れを返す。基準の遅れは直近 10 秒の (復号の出力の時刻 − TIMESTAMP) の最小値、音声の表示の遅れは playout::delay の目標遅延、映像は揺らぎの百分位 (max(0.95, 1 - フレーム間隔 ms / 1000)。上限は既定 500 ms と表示待ちのキューが吸収できる長さの小さい方) である。 observe のたびに観測を足し、present_us で表示時刻を、presentation_delay_us でスケジューラへ渡す音声の遅れを読む。A/V 同期の制御も observe のたびに動く (音声と映像の両方に基準があり、基準の差が閾値の中にあるときだけ)。表示の遅れの差が TIMELINE_SYNC_MIN_DELTA_US を超えたら先行する側へ足して合わせ、足した分は毎秒 TIMELINE_DELAY_DECAY_US_PER_SECOND までで戻す。呼び出し側が別途 sync を呼ぶ必要はない。 targetLatency は TimelineConfig か set_target_latency_ms で渡し、上限に収まらない分は limited_us で読める。TimelineConfig は video_queue_limit (映像の表示待ちのキューの上限、枚)・max_presentation_delay_ms (表示の遅れの上限、既定 500)・target_latency_ms を受ける。(復号の出力の時刻 − TIMESTAMP) が基準から 2 秒以上離れると基準を取り直して generation が進む。

MessageError は Clone 不可。後段へ持ち回る場合は再構築を検討する。

準拠仕様

• draft-ietf-moq-transport-22 - Media over QUIC Transport
  • https://datatracker.ietf.org/doc/html/draft-ietf-moq-transport-22
• draft-ietf-moq-loc-04 - Media over QUIC - Low Overhead Container
  • https://datatracker.ietf.org/doc/html/draft-ietf-moq-loc-04
• draft-ietf-moq-msf-01 - MOQT Streaming Format
  • https://datatracker.ietf.org/doc/html/draft-ietf-moq-msf-01
• draft-ietf-moq-c4m-01 - Authorization scheme for MOQT using Common Access Tokens
  • https://datatracker.ietf.org/doc/html/draft-ietf-moq-c4m-01
• RFC 8949 - Concise Binary Object Representation (CBOR)
  • https://www.rfc-editor.org/rfc/rfc8949
• RFC 9052 / RFC 9053 - CBOR Object Signing and Encryption (COSE)
  • https://www.rfc-editor.org/rfc/rfc9052
• RFC 8392 - CBOR Web Token (CWT)
  • https://www.rfc-editor.org/rfc/rfc8392

注意事項

• 型は定義元モジュールから import する。re-export は存在しない
• Session は sans-I/O の状態機械であり、I/O・非同期処理・MoQ リレーが担う namespace 発見・告知と forwarding は含まない。poll_event() の結果を I/O 層へ渡すのは利用側の責務
• SendRequestError のローカルエラー (PeerGoawayReceived / LocalFilterMismatch / LocalDatagramTimeout) は wire コードを持たない。wire コードとして公開 API に渡す必要はない (渡しても各レジストリの *_INTERNAL_ERROR に置換される)
• ControlMessage と各メッセージ構造体は Clone だが Copy ではない
• MessageError は Clone / Copy 不可
• c4m の署名 / 検証には aws-lc-rs feature が必要。無効時は CatToken::verify / DpopProof::verify_signature などを呼べる実装が存在しない (trait と鍵表現だけがビルドされる)
• c4m のトークン発行は HMAC-SHA256 に RFC 9053 の 5 を使う。draft-ietf-moq-c4m-01 付録 A のベクタが使う -4 は検証でのみ HMAC-SHA256 として受理する
• 開発中のライブラリであり、仕様は積極的に変更される場合がある