ExMoQ publishing & subscribing

Mix.install([
  {:ex_moq, path: Path.join(__DIR__, "..")},
  {:kino, "~> 0.14"},
  {:pythonx, "~> 0.4.2"},
  {:kino_pythonx, "~> 0.1.0"}
])

Logger.configure(level: :info)

[project]
name = "ex_moq_example"
version = "0.0.0"
requires-python = "==3.13.*"
dependencies = ["moq-rs"]

Setup

MoQ uses relays for distribution. This livebook hosts its own to show usage of ex_moq.

import threading, queue, asyncio, moq

startup = queue.SimpleQueue()

def run():
    async def main():
        async with moq.Server("127.0.0.1:0", tls_generate=["localhost"]) as server:
            startup.put(server.local_addr)
            await server.serve()
    asyncio.run(main())

threading.Thread(target=run, daemon=True).start()
relay_addr = startup.get(timeout=10)

Each MoQ track needs to be assigned to a broadcast. A broadcast is a collection of tracks, and a session is a collection of broadcasts. For details, see https://doc.moq.dev/concept/layer/moq-lite.html#terminology

url = "https://#{Pythonx.decode(relay_addr)}"
disable_tls_verify? = true
broadcast = "my_broadcast"
track = "my_video_track"

Publishing

alias ExMoQ.Native

{:ok, pub_session} = Native.create_session(url, self(), disable_tls_verify?)

receive do
  :moq_connected -> :ok
after
  10_000 -> raise "timeout"
end

{:ok, producer} = Native.create_broadcast_producer(pub_session, broadcast)

Track metadata is described using the hang media format. Track formats are advertised through a catalog.json meta-track.

ex_moq NIFs handle operating on the catalog for you.

For MoQ publishers, you just need to ensure the format matches the data you're streaming so subscribers don't get confused.

For subscribers, each change to the catalog.json track is reported as a

{:moq_catalog, broadcast, renditions :: %{track() => track_format() | :unrecognized}}

message sent to the configured process, containing the latest catalog snapshot. Subscribers are expected to react to these changes appropriately, e.g. when the format of a track changes in-place.

alias ExMoQ.WebCodecs

format = %WebCodecs.VideoTrackFormat{
  params: %WebCodecs.VideoTrackParams{width: 1280, height: 720, framerate: 30.0},
  description: <<>>, # empty DCR
  codec: %WebCodecs.H264Codec{in_band: true, profile: 100, constraints: 0, level: 31}
}

:ok =
  Native.add_track(producer, track, format, _priority = nil, :loc, _latency_ns = 0)

frame_duration_ns = div(1_000_000_000, 30)

{:ok, publisher_pid} =
  Kino.start_child(
    {Task,
     fn ->
       Stream.iterate(0, &(&1 + 1))
       |> Stream.each(fn i ->
         keyframe? = rem(i, 30) == 0
         payload = <<i::32, 0::8*100>>

         :ok =
           Native.send_frame(
             producer,
             track,
             i * frame_duration_ns,
             keyframe?,
             payload
           )

         Process.sleep(33)
       end)
       |> Stream.run()
     end}
  )

Subscribing

require Logger

{:ok, subscriber_pid} =
  Kino.start_child({Task,
   fn ->
     {:ok, sub_session} = Native.create_session(url, self(), disable_tls_verify?)

     receive do
       :moq_connected -> :ok
     after
       10_000 -> raise "timeout"
     end

     {:ok, consumer} =
       Native.create_broadcast_consumer(sub_session, broadcast, self(), _latency_ns = 0)

     receive do
       {:moq_catalog, ^broadcast, renditions} when is_map_key(renditions, track) ->
         IO.inspect(renditions[track], label: "format")
     after
       10_000 -> raise "track was not announced"
     end

     # `token` is any integer you choose; it tags this subscription's messages.
     :ok = Native.subscribe_track(consumer, track, _token = 1, _priority = nil)

     handle_frame = fn step, payload, timestamp_ns, keyframe? ->
       IO.puts(
         "[#{step}] received #{byte_size(payload)}-byte frame, timestamp: #{timestamp_ns} ns, keyframe: #{keyframe?}"
       )
     end

     Stream.repeatedly(fn ->
       receive do
         {:moq_frame, 1, <<step::32, _rest::binary>> = payload, timestamp_ns, keyframe?} ->
           handle_frame.(step, payload, timestamp_ns, keyframe?)

         {:moq_track_finished, 1} ->
           Logger.info("track finished")

         {:moq_broadcast_closed, ^broadcast, reason} ->
           Logger.info("broadcast closed: #{inspect(reason)}")

         {:moq_track_error, 1, reason} ->
           raise "subscription failed: #{inspect(reason)}"
       end
     end)
     |> Stream.run()
   end})

Updating a track's format in-place is supported to a limited extent. Changing the media kind is not allowed (e.g. audio -> video).

Native.update_track(producer, track, %{format | params: %{format.params | height: 600}})

Process.exit(publisher_pid, :shutdown)
:ok = Native.remove_track(producer, track)
:ok = Native.close_broadcast_producer(producer)
Process.exit(subscriber_pid, :shutdown)