Exercise ergonomic bindings (Go/Kotlin/Swift) + bump client dev tooling

clients/go/go.mod

@@ -2,4 +2,6 @@ module moqsmoke
 
 go 1.23
 
-require github.com/moq-dev/moq-go v0.2.15
+require github.com/moq-dev/moq-go v0.5.0
+
+require github.com/moq-dev/moq-go-ffi v0.3.2 // indirect

clients/go/smoke.go

@@ -1,16 +1,20 @@
-// Cross-language interop client for the smoke test, built against the published
-// github.com/moq-dev/moq-go module (the raw uniffi-bindgen-go surface).
+// Cross-language interop client for the smoke test, built against the ergonomic
+// github.com/moq-dev/moq-go wrapper (package moq: Dial, CreateBroadcast,
+// PublishMediaStream, SubscribeMedia, range-over-func frame iterators). The raw
+// uniffi-bindgen-go surface lives in github.com/moq-dev/moq-go-ffi; this client
+// exercises the idiomatic wrapper a real Go user would reach for.
 //
-// publish:   read raw Annex-B H.264 from stdin (e.g. piped from ffmpeg) and feed
-//            it to a streaming importer, which infers frame boundaries.
-// subscribe: connect, find the video track in the catalog, and exit 0 as soon as
-//            any non-empty frame arrives (exit 1 on timeout / no data).
+// publish reads raw Annex-B H.264 from stdin (e.g. piped from ffmpeg) and feeds
+// it to a streaming importer, which infers frame boundaries. subscribe connects,
+// finds the video track in the catalog, and exits 0 as soon as any non-empty
+// frame arrives (exit 1 on timeout / no data). Usage:
 //
-//	ffmpeg ... -f h264 - | go-smoke publish   --url http://127.0.0.1:4443 --broadcast b.hang
-//	                       go-smoke subscribe --url http://127.0.0.1:4443 --broadcast b.hang --timeout 20
+//	ffmpeg ... -f h264 - | go-smoke publish --url http://127.0.0.1:4443 --broadcast b.hang
+//	go-smoke subscribe --url http://127.0.0.1:4443 --broadcast b.hang --timeout 20
 package main
 
 import (
+	"context"
 	"flag"
 	"fmt"
 	"io"
@@ -20,10 +24,7 @@ import (
 	"github.com/moq-dev/moq-go/moq"
 )
 
-const (
-	readChunk    = 64 * 1024
-	maxLatencyMs = 1000 // subscribe_media congestion-control / lookahead window
-)
+const readChunk = 64 * 1024
 
 func main() {
 	if len(os.Args) < 2 {
@@ -61,36 +62,28 @@ func main() {
 }
 
 func publish(url, broadcast string) error {
-	origin := moq.NewMoqOriginProducer()
-	defer origin.Destroy()
-
-	producer, err := moq.NewMoqBroadcastProducer()
+	// The relay uses a self-signed cert, so verification is off. With no origin
+	// options, Dial shares one origin for publish + consume.
+	client, err := moq.Dial(context.Background(), url, moq.WithTLSVerify(false))
 	if err != nil {
 		return err
 	}
-	defer producer.Destroy()
+	defer client.Close()
 
-	// avc3: a self-describing Annex-B H.264 stream the importer can frame on its own.
-	media, err := producer.PublishMediaStream("avc3")
+	producer, err := client.CreateBroadcast(broadcast)
 	if err != nil {
 		return err
 	}
-	defer media.Destroy()
-
-	if err := origin.Publish(broadcast, producer); err != nil {
-		return err
-	}
-
-	client := moq.NewMoqClient()
-	defer client.Destroy()
-	client.SetTlsDisableVerify(true)
-	client.SetPublish(&origin)
+	defer producer.Finish()
 
-	session, err := client.Connect(url)
+	// avc3: a self-describing Annex-B H.264 stream the importer can frame on its
+	// own. PublishMediaStream feeds the raw byte stream; whole frames are emitted
+	// as they complete.
+	media, err := producer.PublishMediaStream("avc3")
 	if err != nil {
 		return err
 	}
-	defer session.Destroy()
+	defer media.Finish()
 
 	fmt.Fprintf(os.Stderr, "publishing %q (Annex-B H.264 from stdin) to %s\n", broadcast, url)
 
@@ -113,71 +106,45 @@ func publish(url, broadcast string) error {
 }
 
 func subscribe(url, broadcast string, timeoutS float64) error {
-	done := make(chan error, 1)
-	go func() { done <- subscribeInner(url, broadcast) }()
-
-	select {
-	case err := <-done:
-		return err
-	case <-time.After(time.Duration(timeoutS * float64(time.Second))):
-		// The FFI calls below block; the process exit tears down the goroutine.
-		return fmt.Errorf("timed out waiting for data")
-	}
-}
-
-func subscribeInner(url, broadcast string) error {
-	origin := moq.NewMoqOriginProducer()
-	defer origin.Destroy()
+	// The whole subscribe must complete within the timeout; a cancelled context
+	// aborts any in-flight wrapper call and unblocks the frame iterator.
+	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(timeoutS*float64(time.Second)))
+	defer cancel()
 
-	client := moq.NewMoqClient()
-	defer client.Destroy()
-	client.SetTlsDisableVerify(true)
-	client.SetConsume(&origin)
-
-	session, err := client.Connect(url)
+	client, err := moq.Dial(ctx, url, moq.WithTLSVerify(false))
 	if err != nil {
 		return err
 	}
-	defer session.Destroy()
-
-	consumer := origin.Consume()
-	defer consumer.Destroy()
+	defer client.Close()
 
-	announced, err := consumer.AnnouncedBroadcast(broadcast)
+	announced, err := client.AnnouncedBroadcast(broadcast)
 	if err != nil {
 		return err
 	}
-	defer announced.Destroy()
+	defer announced.Cancel()
 
-	bc, err := announced.Available()
+	bc, err := announced.Available(ctx)
 	if err != nil {
 		return err
 	}
-	defer bc.Destroy()
 
-	name, video, err := videoTrack(bc)
+	name, video, err := videoTrack(ctx, bc)
 	if err != nil {
 		return err
 	}
 
-	media, err := bc.SubscribeMedia(name, video.Container, maxLatencyMs)
+	media, err := bc.SubscribeMedia(name, video.Container, nil)
 	if err != nil {
 		return err
 	}
-	defer media.Destroy()
+	defer media.Cancel()
 
-	total := 0
-	for {
-		frame, err := media.Next()
+	for frame, err := range media.Frames(ctx) {
 		if err != nil {
 			return err
 		}
-		if frame == nil {
-			break
-		}
-		total += len(frame.Payload)
-		if total > 0 {
-			fmt.Fprintf(os.Stderr, "received %d bytes from %q\n", total, broadcast)
+		if len(frame.Payload) > 0 {
+			fmt.Fprintf(os.Stderr, "received %d bytes from %q\n", len(frame.Payload), broadcast)
 			return nil
 		}
 	}
@@ -187,23 +154,20 @@ func subscribeInner(url, broadcast string) error {
 // videoTrack waits for a catalog update that actually carries a video track. A
 // lazy publisher (e.g. the browser, which only encodes on demand) may announce
 // video in a later update, not the first snapshot.
-func videoTrack(bc *moq.MoqBroadcastConsumer) (string, moq.MoqVideo, error) {
+func videoTrack(ctx context.Context, bc *moq.BroadcastConsumer) (string, moq.Video, error) {
 	cat, err := bc.SubscribeCatalog()
 	if err != nil {
-		return "", moq.MoqVideo{}, err
+		return "", moq.Video{}, err
 	}
-	defer cat.Destroy()
+	defer cat.Cancel()
 
-	for {
-		catalog, err := cat.Next()
+	for catalog, err := range cat.Updates(ctx) {
 		if err != nil {
-			return "", moq.MoqVideo{}, err
-		}
-		if catalog == nil {
-			return "", moq.MoqVideo{}, fmt.Errorf("catalog stream ended without a video track")
+			return "", moq.Video{}, err
 		}
 		for name, video := range catalog.Video {
 			return name, video, nil
 		}
 	}
+	return "", moq.Video{}, fmt.Errorf("catalog stream ended without a video track")
 }

clients/js-native/package.json

@@ -6,9 +6,9 @@
     "@moq/hang": "latest",
     "@moq/net": "latest",
     "@moq/web-transport": "latest",
-    "zod": "^4.0.0"
+    "zod": "^4.4.3"
   },
   "devDependencies": {
-    "tsx": "^4.0.0"
+    "tsx": "^4.23.1"
   }
 }

clients/js/package.json

@@ -7,8 +7,8 @@
     "@moq/watch": "latest"
   },
   "devDependencies": {
-    "esbuild": "^0.27.0",
+    "esbuild": "^0.28.1",
     "playwright": "1.59.1",
-    "vite": "^7.3.1"
+    "vite": "^8.1.5"
   }
 }

clients/kotlin/build.gradle.kts

@@ -1,5 +1,5 @@
 plugins {
-    kotlin("jvm") version "2.0.21"
+    kotlin("jvm") version "2.4.10"
     application
 }
 
@@ -10,7 +10,7 @@ dependencies {
     // dependency lockfile is committed, and caches of dynamic versions are
     // disabled below, so each run re-resolves to the newest release.
     implementation("dev.moq:moq:latest.release")
-    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.9.0")
+    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.11.0")
 }
 
 configurations.all {

clients/kotlin/src/main/kotlin/Main.kt

@@ -1,30 +1,28 @@
 // Subscribe-only cross-language interop client for the smoke test, built on the
-// published Kotlin package (dev.moq:moq on Maven Central). Connect, find the
-// video track in the catalog, and exit 0 as soon as any non-empty frame arrives
-// (1 on timeout).
+// ergonomic Kotlin API of the published dev.moq:moq package (the `dev.moq.Moq`
+// connect facade + the `dev.moq` Flow extensions), not the raw uniffi.moq
+// handles. Connect, find the video track in the catalog, and exit 0 as soon as
+// any non-empty frame arrives (1 on timeout).
 //
 //   smoke subscribe --url http://127.0.0.1:4443 --broadcast b.hang --timeout 20
 //
 // Publishing isn't wired up: the raw-stream importer the other clients publish
-// with isn't in the published 0.2.x FFI yet, so this client only subscribes.
+// with isn't exercised here, so this client only subscribes.
+import dev.moq.BroadcastConsumer
+import dev.moq.Moq
+import dev.moq.Video
 import dev.moq.frames
 import dev.moq.updates
 import kotlinx.coroutines.TimeoutCancellationException
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.mapNotNull
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.withTimeout
-import uniffi.moq.MoqBroadcastConsumer
-import uniffi.moq.MoqClient
-import uniffi.moq.MoqOriginProducer
-import uniffi.moq.MoqVideo
 import kotlin.system.exitProcess
 
-private const val MAX_LATENCY_MS = 1_000uL
-
 // A catalog update that actually carries a video track. A lazy publisher may
 // announce video in a later update, not the first snapshot.
-private suspend fun videoTrack(bc: MoqBroadcastConsumer): Pair<String, MoqVideo> =
+private suspend fun videoTrack(bc: BroadcastConsumer): Pair<String, Video> =
     bc.subscribeCatalog().use { catalog ->
         catalog.updates()
             .mapNotNull { it.video.entries.firstOrNull() }
@@ -33,25 +31,18 @@ private suspend fun videoTrack(bc: MoqBroadcastConsumer): Pair<String, MoqVideo>
     }
 
 private suspend fun subscribe(url: String, broadcast: String) {
-    val origin = MoqOriginProducer()
-    val client = MoqClient()
-    client.setTlsDisableVerify(true)
-    client.setConsume(origin)
-
-    val session = client.connect(url)
-    try {
-        val consumer = origin.consume()
-        val announced = consumer.announcedBroadcast(broadcast)
+    // tlsVerify = false: the smoke relay uses a self-signed cert. `use` shuts the
+    // session down (graceful close) on the way out.
+    Moq.connect(url, tlsVerify = false).use { moq ->
+        val announced = moq.announcedBroadcast(broadcast)
         val bc = announced.available()
 
         val (name, video) = videoTrack(bc)
-        val media = bc.subscribeMedia(name, video.container, MAX_LATENCY_MS)
+        val media = bc.subscribeMedia(name, video.container, null)
 
         // Suspends until the first non-empty frame, or throws if the flow ends.
         val frame = media.frames().first { it.payload.isNotEmpty() }
         System.err.println("received ${frame.payload.size} bytes from $broadcast")
-    } finally {
-        session.cancel(0u) // code 0 = graceful close
     }
 }
 

clients/swift/Sources/smoke/main.swift

@@ -1,16 +1,15 @@
 // Subscribe-only cross-language interop client for the smoke test, built on the
-// published Swift package (moq-dev/moq-swift). Connect, find the video track in
-// the catalog, and exit 0 as soon as any non-empty frame arrives (1 on timeout).
+// ergonomic Swift API of the published moq-dev/moq-swift package (the `Moq`
+// module's Client/Session/BroadcastConsumer classes and AsyncSequence streams),
+// not the raw MoqFFI handles. Connect, find the video track in the catalog, and
+// exit 0 as soon as any non-empty frame arrives (1 on timeout).
 //
 //   smoke subscribe --url http://127.0.0.1:4443 --broadcast b.hang --timeout 20
 //
 // Publishing isn't wired up: the raw-stream importer the other clients publish
-// with isn't in the published 0.2.x FFI yet, so this client only subscribes.
+// with isn't exercised here, so this client only subscribes.
 import Foundation
 import Moq
-// The published Moq module re-exports the generated types via plain `import`
-// (not @_exported yet), so name them from MoqFFI directly.
-import MoqFFI
 
 enum SmokeError: Error { case timeout, noVideo, noData }
 
@@ -39,33 +38,32 @@ func parseArgs() -> Args {
 func warn(_ s: String) { FileHandle.standardError.write((s + "\n").data(using: .utf8)!) }
 
 // A catalog update that actually carries a video track. A lazy publisher may
-// announce video in a later update, not the first snapshot.
-func videoTrack(_ bc: MoqBroadcastConsumer) async throws -> (String, MoqVideo) {
-    let catalog = try bc.subscribeCatalog()
-    for try await update in catalog.updates {
+// announce video in a later update, not the first snapshot. CatalogConsumer is
+// an AsyncSequence of catalog snapshots.
+func videoTrack(_ bc: BroadcastConsumer) async throws -> (String, Video) {
+    for try await update in try await bc.subscribeCatalog() {
         if let first = update.video.first { return (first.key, first.value) }
     }
     throw SmokeError.noVideo
 }
 
 func subscribe(_ args: Args) async throws {
-    let origin = MoqOriginProducer()
-    let client = MoqClient()
-    client.setTlsDisableVerify(disable: true)
-    client.setConsume(origin: origin)
+    // The smoke relay uses a self-signed cert. connect(to:) wires an auto-created
+    // origin, so session.consumer discovers announcements without touching MoqFFI.
+    let client = Client()
+    client.setTlsVerify(false)
 
-    let session = try await client.connect(url: args.url)
-    defer { session.cancel(code: 0) } // code 0 = graceful close
+    let session = try await client.connect(to: args.url)
+    defer { session.shutdown() } // graceful close
 
-    let consumer = origin.consume()
-    let announced = try consumer.announcedBroadcast(path: args.broadcast)
+    let announced = try session.consumer.announcedBroadcast(path: args.broadcast)
     let bc = try await announced.available()
 
     let (name, video) = try await videoTrack(bc)
-    let media = try bc.subscribeMedia(name: name, container: video.container, maxLatencyMs: 1000)
+    let media = try await bc.subscribeMedia(name: name, container: video.container)
 
     var total = 0
-    for try await frame in media.frames {
+    for try await frame in media {
         total += frame.payload.count
         if total > 0 {
             warn("received \(total) bytes from \(args.broadcast)")