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

MoQT ハブ

MoQT（Media over QUIC Transport）は、映像や音声などのメディアを QUIC の上で運ぶためのプロトコルです。 そのサーバーは、WebTransport のコールバックの上に置いた wired_moqt_hub です。 接続時に相手が提示する WebTransport のサブプロトコル（moqt-18/moqt-19/moqt-22）に応じて、draft-ietf-moq-transport の 18/19/22 のいずれかを選んで話します。 このガイドの MoQT サンプルのサーバーはすべて opt.wt_protocols = "moqt-22" を設定し、Go のクライアントも moqt-22 を提示するので、ハブは draft-22 で動き、このガイドが引く節番号は draft-22 のものです。 このページではハブだけを動かし、接続したクライアントがサーバーの SETUP（セッションを始める最初のメッセージ）を受け取ります。

コード

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


static wired_moqt_hub g_hub;


int wired_main(int argc, char** argv) {
  /* The same fixed demo identity as the HTTP/3 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-mq");
  id.max_datagram_frame_size = 65535;


  wired_moqt_io io = wired_moqraw_io();
  wired_moqt_init(&g_hub, io);


  /* The hub owns every WebTransport session. A client that offers the
   * subprotocol "moqt-22" gets draft-ietf-moq-transport-22: the hub sends
   * SETUP on a uni control stream of its own (6.3), answers each request
   * on the bidi stream it arrived on (6.4.2), and forgets it on close. */
  wired_srvrun_opt opt     = {0};
  opt.incoming_cpu         = -1;
  opt.wt_protocols         = "moqt-22";
  opt.wt_on_session        = wired_moqt_on_session;
  opt.wt_session_ctx       = &g_hub;
  opt.wt_on_stream_data    = wired_moqt_on_stream_data;
  opt.wt_stream_data_ctx   = &g_hub;
  opt.wt_on_datagram       = wired_moqt_on_datagram;
  opt.wt_datagram_ctx      = &g_hub;
  opt.wt_on_session_close  = wired_moqt_on_session_close;
  opt.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"
  "context"
  "crypto/tls"
  "fmt"
  "io"
  "log"
  "math/bits"
  "net/http"
  "os"
  "time"


  "github.com/quic-go/quic-go"
  "github.com/quic-go/webtransport-go"
)


// varint reads a MoQT variable-length integer: the count of leading 1 bits
// in the first byte is the count of bytes that follow.
func varint(r *bufio.Reader) uint64 {
  b, err := r.ReadByte()
  check(err)
  n := bits.LeadingZeros8(^b)
  v := uint64(b) & (0xff >> (n + 1))
  for ; n > 0; n-- {
    c, err := r.ReadByte()
    check(err)
    v = v<<8 | uint64(c)
  }
  return v
}


// readMsg reads one control message: Type (varint), 16-bit Length, body.
func readMsg(r *bufio.Reader) (uint64, []byte) {
  typ := varint(r)
  var hdr [2]byte
  _, err := io.ReadFull(r, hdr[:])
  check(err)
  body := make([]byte, int(hdr[0])<<8|int(hdr[1]))
  _, err = io.ReadFull(r, body)
  check(err)
  return typ, body
}


func check(err error) {
  if err != nil {
    log.Fatal(err)
  }
}


func main() {
  ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
  defer cancel()
  // Keep the QUIC connection so main can close it before exiting.
  var conn *quic.Conn
  // Offering the WebTransport subprotocol "moqt-22" selects
  // draft-ietf-moq-transport-22 (SS6.2.1).
  d := webtransport.Transport{
    TLSClientConfig:      &tls.Config{InsecureSkipVerify: true},
    QUICConfig:           &quic.Config{EnableDatagrams: true, EnableStreamResetPartialDelivery: true},
    ApplicationProtocols: []string{"moqt-22"},
    DialAddr: func(ctx context.Context, addr string, t *tls.Config, c *quic.Config) (*quic.Conn, error) {
      var err error
      conn, err = quic.DialAddrEarly(ctx, addr, t, c)
      return conn, err
    },
  }
  rsp, s, err := d.Dial(ctx, os.Args[1]+"/moqt", http.Header{})
  check(err)
  fmt.Println("status", rsp.StatusCode)
  if p := s.SessionState().ApplicationProtocol; p != "moqt-22" {
    log.Fatalf("server picked subprotocol %q", p)
  }


  // Each side opens one uni control stream that starts with SETUP
  // (SS6.3). Ours: Type 0x2F00 (also the stream type), Length 0, no
  // Setup Options. It stays open for the whole session.
  ctl, err := s.OpenUniStreamSync(ctx)
  check(err)
  _, err = ctl.Write([]byte{0xAF, 0x00, 0, 0})
  check(err)


  // The server's control stream, and the SETUP on it.
  srv, err := s.AcceptUniStream(ctx)
  check(err)
  typ, body := readMsg(bufio.NewReader(srv))
  fmt.Printf("recv SETUP type=%#x options=%d bytes\n", typ, len(body))
  check(conn.CloseWithError(0, ""))
}

実行結果

Terminal window
$ ./moqt-hub --port 4433
conn closed by peer (CONNECTION_CLOSE)
Terminal window
$ go run . https://127.0.0.1:4433
status 200
recv SETUP type=0x2f00 options=4 bytes

仕組み

Section titled “仕組み”
  • wired_moqt_init(&hub, io) がハブを初期化します。 io は、ハブが使う WebTransport の送信関数の表です。 wired_moqraw_io()（app/moqt/qraw/moqrawio.h）が SDK 既製の表を返します。 この表は、ハブが開くストリームの先頭にストリームシグナル（目印のバイト列）を付け、raw QUIC 上の MoQT セッションにも対応します。
  • ハブの wired_moqt_on_session、_on_stream_data、_on_datagram、_on_session_close は、それぞれ &hub をコンテキストにして wired_srvrun_opt にそのまま設定します。 サーバーを動かすのは wired_server_run_opt です。
  • セッションが開くと、ハブは制御ストリームを開き、その上で SETUP（type 0x2f00。ストリーム種別も兼ねます。draft-ietf-moq-transport-22 の 6.4.1、9.1）を送ります。 このストリームの形は WebTransport のサブプロトコルで決まります。 このページのサーバーは opt.wt_protocols = "moqt-22" を設定し、クライアントも moqt-22（webtransport-go の ApplicationProtocols）を提示するので、ハブが単方向の制御ストリームを開き、クライアントも自分の単方向の制御ストリームを開いて自分の SETUP を送ります（6.3）。 ハブは両方の SETUP が揃ってから要求を処理し、以降の要求（SUBSCRIBE、PUBLISH、FETCH、PUBLISH_NAMESPACE、SUBSCRIBE_NAMESPACE など）は、1 つずつクライアントが開く専用の双方向ストリームに載せ、その上で応答されます（6.4.2）。 クライアントは、サーバーの SETUP を最初に届く単方向ストリーム（AcceptUniStream）から読みます。
  • サブプロトコルを提示しないクライアント（WebTransport の protocols に対応しないブラウザ）には、draft-19 を従来の双方向の制御ストリーム 1 本の上で話す形になり、SETUP は AcceptStream で読みます。 examples/moqt_chat は、これをフォールバックとして使っています。
  • クライアントに版を選ばせるには、wired_moqt_wt_protocols(buf, sizeof buf) が書き出すハブの一覧（moqt-22 moqt-19 moqt-18）を opt.wt_protocols に設定します。 examples/moqt_interop がこの使い方をしています。
  • サーバーの識別情報で max_datagram_frame_size を設定すると、QUIC DATAGRAM（再送しないメッセージを運ぶ QUIC の拡張）が有効になります。 WebTransport のクライアントはこれを必要とします。