過去の Object の取得

途中から参加した購読者は、参加後に公開された Object しか受け取れません。それより前の Object は FETCH で要求し、ハブは Object キャッシュから応答します。このページは小さなキャッシュを取り付け、パブリッシャーにグループ 0 から 5 を送らせます。遅れて参加した視聴者が購読し、過去の分を埋めて（fill）、ライブの購読へ隙間なくつなぎます。単独の FETCH も実行します。

コード

#define WIRED_MAIN
#include "wired.h"

/* Room for exactly 4 cached Objects of the 7-byte payloads the client
 * publishes ("frame-N"): each costs MOQCACHE_HDR on top of its payload.
 * The client publishes 6 groups of one Object each, so the oldest 2 are
 * evicted and a FETCH reaching back to them gets an End of Unknown Range. */
#define FRAME_LEN 7
static u8             g_arena[4 * (MOQCACHE_HDR + FRAME_LEN)];
static wired_moqt_hub g_hub;

static void on_step(void* ctx, u64 now_ms) {
  wired_moqt_tick((wired_moqt_hub*)ctx, now_ms);
}

int wired_main(int argc, char** argv) {
  /* The same fixed demo identity as the other MoQT pages, plus QUIC
   * DATAGRAM support, which WebTransport clients ask for. */
  static wired_srvboot_demo_keys keys;
  wired_srvboot_id               id;
  wired_srvboot_demo(&id, &keys, 0x50, "guide-mf");
  id.max_datagram_frame_size = 65535;

  /* Requests arrive on the client's own bidi streams (draft-ietf-moq-
   * transport-22 6.4.2): stream_reply_open answers on them. A FETCH
   * answer, and the fill a SUBSCRIBE asks for (3.4), use open_uni_stream
   * + stream_send rounds, and stream_reset when one is cancelled. */
  wired_moqt_io io = wired_moqraw_io();
  wired_moqt_init(&g_hub, io);

  /* Every whole Object a publisher sends is now kept for FETCH, oldest
   * whole groups evicted first when the arena is full. */
  wired_moqt_cache_attach(&g_hub, g_arena, sizeof g_arena);

  /* The tick retries a FETCH round the transport refused (and gives up
   * on one that stays refused). "moqt-22" selects draft-22. */
  wired_srvrun_opt opt = {
      .incoming_cpu         = -1,
      .wt_protocols         = "moqt-22",
      .on_step              = on_step,
      .on_step_ctx          = &g_hub,
      .wt_on_session        = wired_moqt_on_session,
      .wt_session_ctx       = &g_hub,
      .wt_on_stream_data    = wired_moqt_on_stream_data,
      .wt_stream_data_ctx   = &g_hub,
      .wt_on_stream_reset   = wired_moqt_on_stream_reset,
      .wt_stream_reset_ctx  = &g_hub,
      .wt_on_session_close  = wired_moqt_on_session_close,
      .wt_session_close_ctx = &g_hub,
  };

  u16 port                 = (u16)wired_cliargs_int(argc, argv, "--port", 4433);
  wired_srvrun_handler h   = {0};
  wired_srvrun_obs     obs = {0};
  return wired_server_run_opt(port, &id, h, obs, &opt) ? 0 : 1;
}

クライアント

package main

import (
  "bufio"
  "context"
  "fmt"
  "log"
  "os"
  "time"

  "wired-guide/moqtclient"
)

// publishGroup sends Group g of track alias 1 as one Object on its own
// uni stream: SUBGROUP_HEADER (Type 0x70: FIRST_OBJECT -- the Original
// Publisher opening a subgroup MUST set it, SS2.2 -- Subgroup 0, default
// priority, no properties), Object ID delta 0, length, "frame-g".
func publishGroup(ctx context.Context, s *moqtclient.Session, g byte) {
  st, err := s.OpenUniStreamSync(ctx)
  moqtclient.Check(err)
  _, err = st.Write(append([]byte{0x70, 1, g, 0, 7}, fmt.Sprintf("frame-%d", g)...))
  moqtclient.Check(err)
  moqtclient.Check(st.Close())
}

// waitIngested returns once the hub holds Group g. A FETCH of Group g is
// answered REQUEST_ERROR (0x5) INVALID_RANGE (0x11) while g is past the
// track's Largest Object (draft-ietf-moq-transport-22 SS3.2), and
// FETCH_OK (0x18) once it is not, so it is retried every 100 ms, up to
// 3 s.
// Request IDs 2, 4, ... follow the PUBLISH's 0 (each fits one byte).
func waitIngested(ctx context.Context, s *moqtclient.Session, g byte) {
  for id := byte(2); id < 62; id += 2 {
    st, r := s.Request(ctx, 0x16, moqtclient.Fetch(id, "clock", g, g))
    // Our side of this request is done: closing it frees the stream,
    // so the retries never run out of stream credit.
    moqtclient.Check(st.Close())
    typ, _ := moqtclient.ReadMsg(r)
    if typ == 0x18 {
      uni, err := s.AcceptUniStream(ctx) // drain this FETCH's Objects
      moqtclient.Check(err)
      moqtclient.ReadFetch(bufio.NewReader(uni))
      return
    }
    time.Sleep(100 * time.Millisecond)
  }
  log.Fatalf("the hub never took in Group %d", g)
}

// printFetch prints every item of the next fetch data stream and returns
// the last one.
func printFetch(ctx context.Context, s *moqtclient.Session, label string) moqtclient.FetchItem {
  uni, err := s.AcceptUniStream(ctx)
  moqtclient.Check(err)
  rid, items := moqtclient.ReadFetch(bufio.NewReader(uni))
  for _, it := range items {
    switch {
    case it.Unknown:
      fmt.Printf("%s (request %d): end of unknown range up to %d/%d\n", label, rid, it.Group, it.Object)
    case it.TimedOut:
      fmt.Printf("%s (request %d): end of timed-out range up to %d/%d\n", label, rid, it.Group, it.Object)
    case it.Range:
      fmt.Printf("%s (request %d): end of non-existent range up to %d/%d\n", label, rid, it.Group, it.Object)
    default:
      fmt.Printf("%s (request %d): %d/%d %q\n", label, rid, it.Group, it.Object, it.Payload)
    }
  }
  return items[len(items)-1]
}

func main() {
  ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
  defer cancel()

  // The publisher: PUBLISH guide/clock (Request ID 0, Track Alias 1),
  // answered by REQUEST_OK (0x7), then Groups 0..5. The hub caches them
  // in its 4-Object arena, so Groups 0 and 1 are evicted.
  pub := moqtclient.Dial(ctx, os.Args[1]+"/moqt")
  defer pub.Conn.CloseWithError(0, "")
  _, r := pub.Request(ctx, 0x1D, append(moqtclient.Track(0, "clock"), 1, 0))
  if typ, _ := moqtclient.ReadMsg(r); typ != 0x7 {
    log.Fatalf("PUBLISH answered with type %#x", typ)
  }
  for g := byte(0); g < 6; g++ {
    publishGroup(ctx, pub, g)
  }
  // The publisher asks the hub itself whether Group 5 has arrived, so
  // the viewer below joins at a known Largest Object.
  waitIngested(ctx, pub, 5)

  // (1) The viewer joins late with the last 3 Groups, the current one
  // included: SUBSCRIBE (Request ID 0) with LOCATION_FILTER Next Object
  // for the live part and a FILL_PARAMETERS whose LOCATION_FILTER
  // Relative Start 3 asks for Groups 3..5 (Largest Group + 1 - 3 up to
  // the Largest Object). SUBSCRIBE_OK's LARGEST_OBJECT is that Largest
  // Object; the live part starts after it.
  view := moqtclient.Dial(ctx, os.Args[1]+"/moqt")
  defer view.Conn.CloseWithError(0, "")
  reply, _ := view.Subscribe(ctx, 0, "clock", moqtclient.JoinBack(3)...)
  if !reply.Ok || !reply.HasLargest {
    log.Fatalf("SUBSCRIBE failed: %+v", reply)
  }
  fmt.Printf("subscribe (request 0): largest object %d/%d\n", reply.LargestGroup, reply.LargestObject)

  // The fill arrives on a fetch stream of its own whose FETCH_HEADER
  // carries the SUBSCRIBE's Request ID; there is no FETCH_OK.
  last := printFetch(ctx, view, "fill")

  // (2) A FETCH (Request ID 2) of Groups 0..2: FETCH_OK on the request
  // stream with its inclusive End Location, the Objects on a uni stream
  // of their own. The evicted Groups 0 and 1 are asked of the publisher,
  // which does not answer: after the hub's budget they come back as one
  // End of Timed-Out Range.
  _, r = view.Request(ctx, 0x16, moqtclient.Fetch(2, "clock", 0, 2))
  eg, eo := moqtclient.ReadFetchOk(r)
  fmt.Printf("standalone fetch (request 2): FETCH_OK end %d/%d\n", eg, eo)
  printFetch(ctx, view, "standalone fetch")

  // The publisher moves on: Groups 6 and 7 reach the viewer live, each on
  // its own uni stream (one at a time here, since separate streams may
  // arrive in either order).
  var firstLive uint64
  for g := byte(6); g < 8; g++ {
    publishGroup(ctx, pub, g)
    uni, err := view.AcceptUniStream(ctx)
    moqtclient.Check(err)
    _, group, payload := moqtclient.ReadObject(bufio.NewReader(uni))
    fmt.Printf("subscription: %d/0 %q\n", group, payload)
    if g == 6 {
      firstLive = group
    }
  }

  // One Object per Group, so the join is seamless when the first live
  // Group is the one right after the fill's last.
  fmt.Printf("join point: fill ends at %d/%d, live starts at %d/0, seamless=%v\n",
    last.Group, last.Object, firstLive, firstLive == last.Group+1)
}

実行結果

$ ./moqt-fetch --port 4433
conn closed by peer (CONNECTION_CLOSE)
conn closed by peer (CONNECTION_CLOSE)

$ go run . https://127.0.0.1:4433
subscribe (request 0): largest object 5/0
fill (request 0): 3/0 "frame-3"
fill (request 0): 4/0 "frame-4"
fill (request 0): 5/0 "frame-5"
standalone fetch (request 2): FETCH_OK end 2/0
standalone fetch (request 2): end of timed-out range up to 1/4611686018427387903
standalone fetch (request 2): 2/0 "frame-2"
subscription: 6/0 "frame-6"
subscription: 7/0 "frame-7"
join point: fill ends at 5/0, live starts at 6/0, seamless=true

仕組み

このサンプルは、WebTransport サブプロトコル moqt-22 で draft-ietf-moq-transport-22 をネゴシエートします（サーバーが opt.wt_protocols を設定します）。両側が単方向の制御ストリームを開いて SETUP（9.1）を送り、ハブは両方の SETUP が揃ってからリクエストを処理します。
wired_moqt_cache_attach は、FETCH 用のアリーナ（キャッシュに使うメモリ領域）をハブに渡します（draft-ietf-moq-transport-22 9.11）。クライアントが PUBLISH したトラックの Object は、完全に届いたものがすべてここへコピーされます。 Object 1 つあたり、ペイロードに加えて MOQCACHE_HDR バイトを使うので、このアリーナには 7 バイトの frame-N がちょうど 4 つ入ります。 6 グループを公開すると、最も古いグループ 0 と 1 がグループ単位で追い出されます。
キャッシュされるのはクライアントが PUBLISH したトラックだけで、ハブ自身の blob やライブトラックは対象外です。そのため Go クライアントは、パブリッシャーと視聴者の 2 つのセッションを開きます。 PUBLISH、SUBSCRIBE、FETCH の各リクエストは、クライアントが開いた新しい双方向ストリーム（6.4.2）で送られ、ハブは stream_reply_open を通じて同じストリームで応答します。
視聴者は LOCATION_FILTER パラメータ（9.20.9）を Next Object（type 0x05。draft-19 では Largest Object）にして SUBSCRIBE するので、現在の Largest より後の Object だけがライブで届きます。 SUBSCRIBE_OK はその Largest（LARGEST_OBJECT、9.20.17、5/0）を返します。
過去の分を取り戻すため、同じ SUBSCRIBE に FILL_PARAMETERS（0x23、9.20.15）を載せます。その内側の LOCATION_FILTER は Relative Start N（type 0x01、ここでは N = 3）で、現在のグループを含む直近 N グループを {Largest.Group + 1 - N, 0} から求めます（9.20.9。3.4、3.5）。 draft-22 に Joining FETCH はなく、fill は購読に相乗りします。ハブはこれをサーバーが開いた単方向ストリームで送り、その FETCH_HEADER（11.4.1）には SUBSCRIBE 自身の Request ID が入り、FETCH_OK はありません。ここではグループ 3 から 5 が届きます（fill (request 0): 3/0 から 5/0）。最初のライブの Object は 6/0 なので、fill と購読は隙間も重複もなくつながります。
グループ 0 から 2 を求める単独の FETCH は、本体が Request ID、Track Namespace、Track Name、Parameters で、範囲は LOCATION_FILTER パラメータ（type 0x03 の Absolute Start, Group End。開始 0/0、End Group Delta 2）で指定し、Fetch Type フィールドはありません。リクエストストリームで FETCH_OK（9.12）を受け取り、End Location は 2/0（FETCH_OK end 2/0）です。 End Location は draft-22 では範囲に含まれる最後の位置で、draft-19 では最後の Object の 1 つ先でした。 Object は新しい単方向ストリーム（FETCH_HEADER と、それに続く fetch Object）で受け取ります。追い出されたグループ 0 と 1 はキャッシュにないので、ハブはトラックのパブリッシャーに自分から FETCH を送って取りに行きます（3.2.3）。このサンプルのパブリッシャーは FETCH に応答しません。そのためハブは待ち時間の上限（WIRED_MOQTRUN_UPFETCH_BUDGET_MS の 2 秒。FILL_TIMEOUT のほうが短ければそちら）が過ぎるとこの FETCH を取り消し、グループ 0 と 1 を 1/4611686018427387903 までの 1 つの End of Timed-Out Range（0x20C）として返します（2^62−1 は、ハブが「グループ 1 の残り全体」を表すのに使う値です。draft-22 の Object ID は 2^64−1 まで取り得ます）。その後に、キャッシュにあるグループ 2 が続きます。パブリッシャーが応答すれば、ここにはその Object がそのまま届きます。 0x20C がない draft-19 の視聴者には、同じ欠けを End of Unknown Range（0x10C）で返します。
fetch ストリームと fill ストリームは open_uni_stream、stream_send、stream_reset の io エントリで送られます。このストリームは複数ラウンドにわたって開いたままで、トランスポートが拒否したラウンドは、on_step から呼ぶ wired_moqt_tick が再試行します。
視聴者が参加する前に、パブリッシャーはグループ 5 がハブに取り込まれたことを FETCH で確かめます。グループ 5 を求める単独の FETCH は、それがトラックの Largest Object より先にある間は REQUEST_ERROR INVALID_RANGE で、ハブが取り込んだ後は FETCH_OK で応答されます。クライアントは 100 ms ごとに最大 3 秒まで再試行し、リクエストストリームの自分側を毎回閉じるので、再試行でストリームのクレジット（開けるストリーム数の枠）を使い切ることはありません。そのため視聴者は常に Largest が 5/0 の時点で参加します。