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

データグラムとストリームの配送

MoQT の Object（MoQT がデータを運ぶ単位）は、保証の異なる 2 種類のワイヤー形式で運べます。 このページは、1 つの Object を QUIC DATAGRAM として送り、もう少し大きな Object をストリームとして送ります。 データグラムにはロスがあり、再送はなく、コネクション自体のデータグラムサイズに制限されます。 ストリームには信頼性があり、Object 1 つあたり 16 KiB の上限で分割されます。

コード

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


/* Bigger than MOQDATA_BLOB_CHUNK (16384), so the stream delivery below
 * demonstrates the per-Object cap by arriving as two Objects, not one. */
#define BLOB_LEN 20000
static u8 g_blob[BLOB_LEN];
static u8 g_wire[MOQDATA_BLOB_WIRE_CAP(BLOB_LEN)];


/* Any datagram from the client is just a "go ahead" trigger (content
 * unchecked): send one lossy OBJECT_DATAGRAM (draft-ietf-moq-transport-22
 * 11.2.1), then one reliable Object stream (11.3.1) whose blob exceeds the
 * 16 KiB per-Object cap and so arrives chunked into two Objects. */
static void on_datagram(void* ctx, wired_wt_session* s, wired_span data) {
  (void)ctx;
  (void)data;


  u8        dg[32];
  usz       off = 0;
  moqdg_obj o   = {0};
  o.track_alias = 2;
  o.payload     = wired_span_cstr("ping-obj");
  if (moqdg_put(wired_mspan_of(dg, sizeof dg), &off, &o) == MOQDATA_OK)
    wired_server_wt_send_datagram_to(s, wired_span_of(dg, off));


  usz n = moqdata_blob_build(
      wired_mspan_of(g_wire, sizeof g_wire), 1,
      wired_span_of(g_blob, BLOB_LEN));
  if (n) wired_moqraw_io().send_uni(s, wired_span_of(g_wire, n));
}


int wired_main(int argc, char** argv) {
  /* The same fixed demo identity as the HTTP/3 hello page. */
  static wired_srvboot_demo_keys keys;
  wired_srvboot_id               id;
  wired_srvboot_demo(&id, &keys, 0x50, "guide-md");
  id.max_datagram_frame_size = 65535; /* WebTransport requires DATAGRAM */


  for (usz i = 0; i < BLOB_LEN; i++) g_blob[i] = (u8)('a' + i % 26);


  /* No hub here, so no SETUP or requests: the "moqt-22" subprotocol only
   * says the session speaks draft-22's data plane. */
  wired_srvrun_opt opt = {0};
  opt.incoming_cpu     = -1;
  opt.wt_protocols     = "moqt-22";
  opt.wt_on_datagram   = on_datagram;


  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"
  "io"
  "os"
  "time"


  "wired-guide/moqtclient"
)


func main() {
  ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
  defer cancel()
  // A "moqt-22" WebTransport session; this server runs no hub, so no
  // SETUP or request follows -- only the data plane.
  s, conn, status := moqtclient.DialWT(ctx, os.Args[1]+"/deliver")
  fmt.Println("status", status)


  // A "go" datagram tells the server to send one of each delivery mode.
  moqtclient.Check(s.SendDatagram([]byte("go")))


  // Lossy: one OBJECT_DATAGRAM, bounded by the connection's own QUIC
  // datagram size, dropped (never retried) if the network loses it.
  dg, err := s.ReceiveDatagram(ctx)
  moqtclient.Check(err)
  dr := bufio.NewReader(bytes.NewReader(dg))
  moqtclient.Varint(dr) // Type 0x00: Object ID and priority present, no properties, payload
  alias := moqtclient.Varint(dr)
  group := moqtclient.Varint(dr)
  objID := moqtclient.Varint(dr)
  _, err = dr.ReadByte() // raw Publisher Priority byte, not a varint
  moqtclient.Check(err)
  payload, err := io.ReadAll(dr)
  moqtclient.Check(err)
  fmt.Printf("datagram alias=%d group=%d id=%d payload=%q\n", alias, group, objID, payload)


  // Reliable: one Object stream, its 20000-byte blob split at the 16 KiB
  // per-Object cap into two Objects that arrive complete, in order.
  uni, err := s.AcceptUniStream(ctx)
  moqtclient.Check(err)
  ur := bufio.NewReader(uni)
  salias, sgroup, obj1 := moqtclient.ReadObject(ur)
  moqtclient.Varint(ur) // second Object's ID delta
  obj2 := make([]byte, moqtclient.Varint(ur))
  _, err = io.ReadFull(ur, obj2)
  moqtclient.Check(err)
  fmt.Printf("stream alias=%d group=%d object_1=%d bytes object_2=%d bytes total=%d bytes\n",
    salias, sgroup, len(obj1), len(obj2), len(obj1)+len(obj2))


  conn.CloseWithError(0, "")
}

実行結果

Terminal window
$ ./moqt-delivery --port 4433
conn closed by peer (CONNECTION_CLOSE)
Terminal window
$ go run . https://127.0.0.1:4433
status 200
datagram alias=2 group=0 id=0 payload="ping-obj"
stream alias=1 group=0 object_1=16384 bytes object_2=3616 bytes total=20000 bytes

仕組み

Section titled “仕組み”
  • moqdg_put は、OBJECT_DATAGRAM（draft-ietf-moq-transport-22 11.2.1）を直接エンコードします。 wired_server_wt_send_datagram_to は、それを QUIC DATAGRAM としてキューに積むだけで、ストリームも再送も関与しません。
  • moqdata_blob_build は、blob を 1 つの SUBGROUP_HEADER と、それに続く最大 MOQDATA_BLOB_CHUNK（16384）バイトずつの Object へフレーム化します（draft-ietf-moq-transport-22 11.3.1）。 このページの blob は 20000 バイトなので、16384 バイトと残り 3616 バイトの、2 つの Object になります。 wired_server_wt_open_uni は、フレーム化したストリーム全体を FIN まで含めて、信頼性のある形で送ります。
  • クライアントの go run の出力には、両方が現れます。 データグラムは 1 つのペイロードとしてそのまま届き、ストリームは object_1 と object_2 として届き、両者の長さを足すと元の 20000 バイトに戻ります。
  • このページは、moqdata/moqdg を使って Object を直接組み立てており、wired_moqt_hub は経由しません。 PUBLISH / SUBSCRIBE という制御メッセージのやり取り（コントロールプレーン）はなく、ハブ自身も土台にしている 2 つの WebTransport 配送プリミティブだけを使っています。 ハブ自身のより高レベルな API については、トラックの公開 を参照してください。