Skip to content
wired guide

Namespace discovery

A subscriber does not have to know track names in advance. It can watch a namespace prefix and let the hub tell it which namespaces exist under it, as publishers come and go. This page runs one watcher on the prefix guide and four publisher announcements: one before the watcher, two after it, and one the server’s policy refuses. Then one publisher leaves.

Code

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

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)
}

Output

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

How it works

Section titled “How it works”
  • The sample negotiates draft-ietf-moq-transport-22 through the WebTransport subprotocol moqt-22 (the server sets opt.wt_protocols). Each side opens its own unidirectional control stream and sends SETUP (9.1) on it, and the hub serves requests only after both SETUPs have arrived.
  • Each request then goes on a fresh bidi stream the client opens (draft-ietf-moq-transport-22 6.4.2), and the hub answers on that stream through stream_reply_open. Both namespace requests stay in force after their answer, until the stream is cancelled (6.4.2.3) or the session ends. NAMESPACE and NAMESPACE_DONE are pushed on the SUBSCRIBE_NAMESPACE stream itself.
  • PUBLISH_NAMESPACE (9.14, 4.1) announces a Track Namespace (guide/lobby, guide/room-a, …), and the hub answers REQUEST_OK.
  • SUBSCRIBE_NAMESPACE (9.15, 4.2) names a Track Namespace Prefix. After REQUEST_OK the hub sends one NAMESPACE (9.16) for every namespace already published under the prefix (the initial set, here lobby), then one more each time a matching namespace is published. Each carries only the suffix: the fields after the prefix.
  • hub.authorize_namespace sees every PUBLISH_NAMESPACE and SUBSCRIBE_NAMESPACE (its msg_type says which) with the namespace and the presented AUTHORIZATION TOKEN. This sample refuses any namespace ending in secret: the publisher gets REQUEST_ERROR UNAUTHORIZED (1) and the watcher is never told about it.
  • When room-a closes its session, wired_moqt_on_session_close withdraws every namespace it published, and the hub sends NAMESPACE_DONE (9.17) to each watcher that was told about it.
  • The client never sleeps to wait for the hub: each publisher waits for its own REQUEST_OK, and the watcher reads the NAMESPACE or NAMESPACE_DONE the hub pushes, so every step is ordered by messages on the wire. The hub’s tick (wired_moqt_tick from on_step) retries a push the transport refused.