コンテンツにスキップ
wired guide

名前空間の発見

購読者は、トラック名を前もって知っている必要はありません。 名前空間のプレフィックス（名前空間の先頭部分）を監視すれば、その下にどの名前空間があるかを、パブリッシャーの出入りに合わせてハブが知らせます。 このページでは、プレフィックス guide を監視する 1 つのウォッチャーと、4 つの名前空間の公開を動かします。 公開のうち 1 つはウォッチャーの開始より前、2 つは後に行われ、残る 1 つはサーバーのポリシーに拒否されます。 最後に、パブリッシャーの 1 つが退出します。

コード

main.c
#define WIRED_MAIN
#include "wired.h"


/* draft-ietf-moq-transport-22 9.14 / 9.15: the hub asks this hook before
 * it accepts a PUBLISH_NAMESPACE or SUBSCRIBE_NAMESPACE. This sample's
 * policy refuses any namespace ending in "secret" (a real deployment
 * would verify the token too; this demo ignores it). */
static int authorize_ns(
    void* ctx, u64 msg_type, const moqctl_ns* ns, const moqctl_token* token) {
  (void)ctx;
  (void)msg_type;
  (void)token;
  return ns->n == 0 || !wired_span_eq_cstr(ns->fields[ns->n - 1], "secret");
}


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-nd");
  id.max_datagram_frame_size = 65535;


  /* PUBLISH_NAMESPACE and SUBSCRIBE_NAMESPACE arrive on the client's own
   * bidi streams (draft-ietf-moq-transport-22 6.4.2) and stay open: the hub
   * answers on them through stream_reply_open, and later pushes NAMESPACE /
   * NAMESPACE_DONE onto the subscriber's stream with stream_send. */
  wired_moqt_io io = wired_moqraw_io();
  wired_moqt_init(&g_hub, io);
  g_hub.authorize_namespace = authorize_ns;


  /* on_session_close withdraws every namespace a leaving session
   * published (NAMESPACE_DONE to its watchers); the tick retries a push
   * the transport refused. "moqt-22" selects draft-22 (see the hub page). */
  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;
}

クライアント

client.go
package main


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


  "wired-guide/moqtclient"
)


// request sends one request on a fresh bidi request stream
// (draft-ietf-moq-transport-22 SS6.4.2) and prints the hub's first
// answer: REQUEST_OK (0x7) or REQUEST_ERROR (0x5) with its error code.
// The stream is left alone: resetting it would cancel the request.
func request(ctx context.Context, s *moqtclient.Session, who string, typ byte, id byte, name string) *bufio.Reader {
  label := map[byte]string{0x6: "PUBLISH_NAMESPACE", 0x50: "SUBSCRIBE_NAMESPACE"}[typ]
  // Request ID, Track Namespace (or Prefix), no parameters.
  _, r := s.Request(ctx, typ, append(append([]byte{id}, moqtclient.Namespace(strings.Split(name, "/")...)...), 0))
  switch t, body := moqtclient.ReadMsg(r); t {
  case 0x7:
    fmt.Printf("%s: %s %s -> REQUEST_OK\n", who, label, name)
  case 0x5:
    code := moqtclient.Varint(bufio.NewReader(bytes.NewReader(body)))
    fmt.Printf("%s: %s %s -> REQUEST_ERROR error_code=%d\n", who, label, name, code)
  default:
    log.Fatalf("%s answered with type %#x", label, t)
  }
  return r
}


// publish opens a publisher session (one MoQT peer; closing its QUIC
// connection is how it leaves) that announces name (Request ID 0).
func publish(ctx context.Context, who, name string) *moqtclient.Session {
  s := moqtclient.Dial(ctx, os.Args[1]+"/moqt")
  request(ctx, s, who, 0x6, 0, name)
  return s
}


// watch reads the next push on the SUBSCRIBE_NAMESPACE stream and prints
// it with the namespace suffix (the fields after the prefix "guide").
func watch(r *bufio.Reader) {
  typ, body := moqtclient.ReadMsg(r)
  label := map[uint64]string{0x8: "NAMESPACE", 0xE: "NAMESPACE_DONE"}[typ]
  if label == "" {
    log.Fatalf("unexpected push type %#x", typ)
  }
  fmt.Printf("watcher: %s %s\n", label, strings.Join(moqtclient.ReadNamespace(body), "/"))
}


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


  // A namespace already published before anyone watches.
  lobby := publish(ctx, "lobby", "guide/lobby")
  defer lobby.Conn.CloseWithError(0, "")


  // The watcher subscribes to the prefix "guide". After REQUEST_OK the
  // hub sends one NAMESPACE per namespace already published under it:
  // the initial set.
  w := moqtclient.Dial(ctx, os.Args[1]+"/moqt")
  defer w.Conn.CloseWithError(0, "")
  watcher := request(ctx, w, "watcher", 0x50, 0, "guide")
  watch(watcher)


  // Two publishers join. Each waits for its own REQUEST_OK, and the
  // watcher then reads the NAMESPACE the hub pushed for it, so every
  // step is ordered by messages on the wire, never by a sleep.
  roomA := publish(ctx, "room-a", "guide/room-a")
  watch(watcher)
  roomB := publish(ctx, "room-b", "guide/room-b")
  defer roomB.Conn.CloseWithError(0, "")
  watch(watcher)


  // The server's authorize_namespace hook refuses "secret": the
  // publisher gets REQUEST_ERROR UNAUTHORIZED (0x1) and the watcher is
  // never told about it (the next push it sees is room-a leaving).
  request(ctx, roomB, "room-b", 0x6, 2, "guide/secret")


  // room-a leaves: closing its session withdraws its namespace, and the
  // hub pushes NAMESPACE_DONE to the watcher.
  moqtclient.Check(roomA.Conn.CloseWithError(0, ""))
  fmt.Println("room-a: session closed")
  watch(watcher)
}

実行結果

Terminal window
$ ./moqt-discovery --port 4433
conn closed by peer (CONNECTION_CLOSE)
conn closed by peer (CONNECTION_CLOSE)
conn closed by peer (CONNECTION_CLOSE)
conn closed by peer (CONNECTION_CLOSE)
Terminal window
$ go run . https://127.0.0.1:4433
lobby: PUBLISH_NAMESPACE guide/lobby -> REQUEST_OK
watcher: SUBSCRIBE_NAMESPACE guide -> REQUEST_OK
watcher: NAMESPACE lobby
room-a: PUBLISH_NAMESPACE guide/room-a -> REQUEST_OK
watcher: NAMESPACE room-a
room-b: PUBLISH_NAMESPACE guide/room-b -> REQUEST_OK
watcher: NAMESPACE room-b
room-b: PUBLISH_NAMESPACE guide/secret -> REQUEST_ERROR error_code=1
room-a: session closed
watcher: NAMESPACE_DONE room-a

仕組み

Section titled “仕組み”
  • このサンプルは、WebTransport サブプロトコル moqt-22 で draft-ietf-moq-transport-22 をネゴシエートします（サーバーが opt.wt_protocols を設定します）。 両側がそれぞれ自分の単方向制御ストリームを開いて SETUP（9.1）を送り、ハブは両方の SETUP が揃ってからリクエストを処理します。
  • 各リクエストは、クライアントが開いた新しい双方向ストリームで送られます（draft-ietf-moq-transport-22 6.4.2）。 ハブは stream_reply_open を通じて同じストリームで応答します。 名前空間の 2 種類のリクエストは、応答の後も、ストリームが取り消されるか（6.4.2.3）セッションが終わるまで有効です。 NAMESPACE と NAMESPACE_DONE は、SUBSCRIBE_NAMESPACE のストリーム自身に送られます。
  • PUBLISH_NAMESPACE（9.14、4.1）は、Track Namespace（guide/lobby、guide/room-a など）を公開し、ハブは REQUEST_OK で応答します。
  • SUBSCRIBE_NAMESPACE（9.15、4.2）は、Track Namespace Prefix を指定します。 REQUEST_OK の後、ハブはプレフィックスの下で公開済みの名前空間ごとに NAMESPACE（9.16）を 1 つ送ります（初期集合で、ここでは lobby）。 その後も、一致する名前空間が公開されるたびに NAMESPACE を送ります。 どの NAMESPACE も、プレフィックスより後ろのフィールドであるサフィックスだけを運びます。
  • hub.authorize_namespace は、すべての PUBLISH_NAMESPACE と SUBSCRIBE_NAMESPACE を、名前空間と提示された AUTHORIZATION TOKEN とともに受け取ります。 どちらのメッセージかは msg_type でわかります。 このサンプルは、最後のフィールドが secret の名前空間を拒否します。 パブリッシャーには REQUEST_ERROR UNAUTHORIZED（1）が返り、ウォッチャーにはその名前空間が知らされません。
  • room-a がセッションを閉じると、wired_moqt_on_session_close がそのセッションの公開した名前空間をすべて取り下げます。 ハブは、その名前空間を知らせていたウォッチャーそれぞれに NAMESPACE_DONE（9.17）を送ります。
  • クライアントは、ハブを待つために sleep しません。 各パブリッシャーは自分への REQUEST_OK を待ち、ウォッチャーはハブが送ってくる NAMESPACE または NAMESPACE_DONE を読みます。 そのため、各段階の順序はワイヤ上のメッセージで決まります。 トランスポートが拒否した送信は、on_step から呼ぶハブのティック（wired_moqt_tick）が再試行します。