refactor(api): move convenience flows into MOQX.Helpers

Author: dmorn

lib/mix/tasks/moqx.moqtail.demo.ex

@@ -234,8 +234,8 @@ defmodule Mix.Tasks.Moqx.Moqtail.Demo do
   defp catalog_fetch_fallback_reason?(_reason), do: false
 
   defp fetch_catalog(subscriber, namespace, timeout) do
-    with {:ok, ref} <- MOQX.fetch_catalog(subscriber, namespace: namespace),
-         {:ok, catalog} <- MOQX.await_catalog(ref, timeout) do
+    with {:ok, ref} <- MOQX.Helpers.fetch_catalog(subscriber, namespace: namespace),
+         {:ok, catalog} <- MOQX.Helpers.await_catalog(ref, timeout) do
       {:ok, catalog}
     else
       {:error, reason} -> {:error, reason}

lib/moqx.ex

@@ -18,8 +18,9 @@ defmodule MOQX do
     - request-level failures (`{:moqx_request_error, %MOQX.RequestError{}}`)
     - transport/runtime failures (`{:moqx_transport_error, %MOQX.TransportError{}}`)
 
-  Convenience helpers (`write_frame/2`, `publish_catalog/2`, `fetch_catalog/2`,
-  `await_catalog/2`) are wrappers on top of the core primitives.
+  Optional convenience flows (`publish_catalog/2`, `update_catalog/2`,
+  `fetch_catalog/2`, `await_catalog/2`, `await_track_active/2`, etc.) live in
+  `MOQX.Helpers` on top of this core API.
 
   ## Example
 
@@ -376,34 +377,6 @@ defmodule MOQX do
     end
   end
 
-  @doc """
-  Creates and publishes the initial catalog object on the `"catalog"` track.
-
-  This convenience helper composes `create_track/2` and `update_catalog/2`:
-
-      {:ok, catalog_track} = MOQX.publish_catalog(broadcast, catalog_json)
-
-  Returns `{:ok, catalog_track}` when both steps succeed.
-  """
-  @spec publish_catalog(broadcast(), catalog_payload()) ::
-          {:ok, track()} | {:error, MOQX.RequestError.t()}
-  def publish_catalog(broadcast, catalog_payload) when is_binary(catalog_payload) do
-    with {:ok, catalog_track} <- create_track(broadcast, "catalog"),
-         :ok <- update_catalog(catalog_track, catalog_payload) do
-      {:ok, catalog_track}
-    end
-  end
-
-  @doc """
-  Writes one catalog object to an existing catalog track.
-
-  Use this to push catalog updates after the initial `publish_catalog/2` call.
-  """
-  @spec update_catalog(track(), catalog_payload()) :: :ok | {:error, MOQX.RequestError.t()}
-  def update_catalog(track, catalog_payload) when is_binary(catalog_payload) do
-    write_frame(track, catalog_payload)
-  end
-
   @doc """
   Writes one frame to a track.
 
@@ -941,76 +914,6 @@ defmodule MOQX do
     end
   end
 
-  @doc """
-  Fetches the raw catalog track bytes.
-
-  This is a thin wrapper over `fetch/4` with catalog defaults:
-
-  - namespace: `"moqtail"`
-  - track name: `"catalog"`
-  - priority: `0`
-  - group order: `:original`
-  - start: `{0, 0}`
-  - end: `{0, 1}`
-  """
-  @spec fetch_catalog(session(), Keyword.t()) ::
-          {:ok, fetch_ref()} | {:error, MOQX.RequestError.t()}
-  def fetch_catalog(session, opts \\ []) when is_list(opts) do
-    namespace = opts |> Keyword.get(:namespace, "moqtail") |> normalize_fetch_namespace!()
-
-    fetch_opts =
-      opts
-      |> Keyword.delete(:namespace)
-      |> Keyword.put_new(:priority, 0)
-      |> Keyword.put_new(:group_order, :original)
-      |> Keyword.put_new(:start, {0, 0})
-      |> Keyword.put_new(:end, {0, 1})
-
-    fetch(session, namespace, "catalog", fetch_opts)
-  end
-
-  @doc """
-  Collects fetch messages for `ref` and decodes the payload as a CMSF catalog.
-
-  Blocks the caller until all objects are received, then concatenates the
-  payloads and passes them to `MOQX.Catalog.decode/1`.
-
-  Returns `{:ok, catalog}` on success, `{:error, reason}` on fetch failure
-  or decode failure, and `{:error, "timeout"}` if no terminal message arrives
-  within `timeout` milliseconds.
-
-  ## Example
-
-      {:ok, ref} = MOQX.fetch_catalog(subscriber, namespace: "moqtail")
-      {:ok, catalog} = MOQX.await_catalog(ref)
-  """
-  @spec await_catalog(fetch_ref(), timeout()) ::
-          {:ok, MOQX.Catalog.t()} | {:error, String.t()}
-  def await_catalog(ref, timeout \\ 5_000) when is_reference(ref) do
-    await_catalog_loop(ref, [], timeout)
-  end
-
-  defp await_catalog_loop(ref, acc, timeout) do
-    receive do
-      {:moqx_fetch_ok, %MOQX.FetchOk{ref: ^ref}} ->
-        await_catalog_loop(ref, acc, timeout)
-
-      {:moqx_fetch_object, %MOQX.FetchObject{ref: ^ref, payload: payload}} ->
-        await_catalog_loop(ref, [acc | [payload]], timeout)
-
-      {:moqx_fetch_done, %MOQX.FetchDone{ref: ^ref}} ->
-        MOQX.Catalog.decode(IO.iodata_to_binary(acc))
-
-      {:moqx_request_error, %MOQX.RequestError{op: :fetch, ref: ^ref, message: reason}} ->
-        {:error, reason}
-
-      {:moqx_transport_error, %MOQX.TransportError{op: :fetch, ref: ^ref, message: reason}} ->
-        {:error, reason}
-    after
-      timeout -> {:error, "timeout"}
-    end
-  end
-
   defp validate_fetch_opts_keys!(opts) do
     allowed_keys = [:priority, :group_order, :start, :end]
 
@@ -1020,13 +923,6 @@ defmodule MOQX do
     end
   end
 
-  defp normalize_fetch_namespace!(namespace) when is_binary(namespace), do: namespace
-
-  defp normalize_fetch_namespace!(namespace) do
-    raise ArgumentError,
-          "expected catalog :namespace to be a string, got: #{inspect(namespace)}"
-  end
-
   defp normalize_fetch_priority!(priority) when is_integer(priority) and priority in 0..255,
     do: priority
 

lib/moqx/catalog.ex

@@ -2,13 +2,13 @@ defmodule MOQX.Catalog do
   @moduledoc """
   Decodes CMSF catalog payloads and provides track discovery helpers.
 
-  A CMSF catalog is a UTF-8 JSON payload retrieved via `MOQX.fetch_catalog/2`.
+  A CMSF catalog is a UTF-8 JSON payload retrieved via `MOQX.Helpers.fetch_catalog/2`.
   This module parses those raw bytes into an inspectable Elixir structure.
 
   ## Example
 
-      {:ok, ref} = MOQX.fetch_catalog(subscriber, namespace: "moqtail")
-      {:ok, catalog} = MOQX.await_catalog(ref, 5_000)
+      {:ok, ref} = MOQX.Helpers.fetch_catalog(subscriber, namespace: "moqtail")
+      {:ok, catalog} = MOQX.Helpers.await_catalog(ref, 5_000)
 
       catalog
       |> MOQX.Catalog.video_tracks()

lib/moqx/helpers.ex

@@ -0,0 +1,128 @@
+defmodule MOQX.Helpers do
+  @moduledoc """
+  Optional convenience helpers built on top of the low-level `MOQX` core API.
+
+  This module composes core primitives and typed lifecycle events; it does not
+  change core message contracts.
+  """
+
+  @typedoc "Raw CMSF catalog payload bytes (UTF-8 JSON)."
+  @type catalog_payload :: MOQX.catalog_payload()
+
+  @doc """
+  Creates and publishes the initial catalog object on the `"catalog"` track.
+  """
+  @spec publish_catalog(MOQX.broadcast(), catalog_payload()) ::
+          {:ok, MOQX.track()} | {:error, MOQX.RequestError.t()}
+  def publish_catalog(broadcast, catalog_payload) when is_binary(catalog_payload) do
+    with {:ok, catalog_track} <- MOQX.create_track(broadcast, "catalog"),
+         :ok <- update_catalog(catalog_track, catalog_payload) do
+      {:ok, catalog_track}
+    end
+  end
+
+  @doc """
+  Writes one catalog object to an existing catalog track.
+  """
+  @spec update_catalog(MOQX.track(), catalog_payload()) :: :ok | {:error, MOQX.RequestError.t()}
+  def update_catalog(track, catalog_payload) when is_binary(catalog_payload) do
+    MOQX.write_frame(track, catalog_payload)
+  end
+
+  @doc """
+  Fetches the raw catalog track bytes.
+
+  Thin wrapper over `MOQX.fetch/4` with catalog defaults.
+  """
+  @spec fetch_catalog(MOQX.session(), Keyword.t()) ::
+          {:ok, MOQX.fetch_ref()} | {:error, MOQX.RequestError.t()}
+  def fetch_catalog(session, opts \\ []) when is_list(opts) do
+    namespace = opts |> Keyword.get(:namespace, "moqtail") |> normalize_namespace!()
+
+    fetch_opts =
+      opts
+      |> Keyword.delete(:namespace)
+      |> Keyword.put_new(:priority, 0)
+      |> Keyword.put_new(:group_order, :original)
+      |> Keyword.put_new(:start, {0, 0})
+      |> Keyword.put_new(:end, {0, 1})
+
+    MOQX.fetch(session, namespace, "catalog", fetch_opts)
+  end
+
+  @doc """
+  Collects fetch messages for `ref` and decodes the payload as a CMSF catalog.
+  """
+  @spec await_catalog(MOQX.fetch_ref(), timeout()) ::
+          {:ok, MOQX.Catalog.t()} | {:error, String.t()}
+  def await_catalog(ref, timeout \\ 5_000) when is_reference(ref) do
+    await_catalog_loop(ref, [], timeout)
+  end
+
+  @doc """
+  Waits until the given publisher track becomes active.
+
+  Returns `:ok` on activation, `{:error, :timeout}` if no lifecycle event
+  arrives within `timeout`, or `{:error, %MOQX.RequestError{code: :track_closed}}`
+  if the track closes first.
+  """
+  @spec await_track_active(MOQX.track(), timeout()) ::
+          :ok | {:error, :timeout | MOQX.RequestError.t()}
+  def await_track_active(track, timeout \\ 5_000) when is_reference(track) do
+    receive do
+      {:moqx_track_active, %MOQX.TrackActive{track: ^track}} ->
+        :ok
+
+      {:moqx_track_closed, %MOQX.TrackClosed{track: ^track}} ->
+        {:error,
+         %MOQX.RequestError{
+           op: :write_frame,
+           code: :track_closed,
+           message: "track_closed",
+           handle: track
+         }}
+    after
+      timeout -> {:error, :timeout}
+    end
+  end
+
+  @doc """
+  Waits for track activation, then writes one frame.
+  """
+  @spec write_frame_when_active(MOQX.track(), binary(), timeout()) ::
+          :ok | {:error, :timeout | MOQX.RequestError.t()}
+  def write_frame_when_active(track, payload, timeout \\ 5_000)
+      when is_reference(track) and is_binary(payload) do
+    with :ok <- await_track_active(track, timeout) do
+      MOQX.write_frame(track, payload)
+    end
+  end
+
+  defp await_catalog_loop(ref, acc, timeout) do
+    receive do
+      {:moqx_fetch_ok, %MOQX.FetchOk{ref: ^ref}} ->
+        await_catalog_loop(ref, acc, timeout)
+
+      {:moqx_fetch_object, %MOQX.FetchObject{ref: ^ref, payload: payload}} ->
+        await_catalog_loop(ref, [acc | [payload]], timeout)
+
+      {:moqx_fetch_done, %MOQX.FetchDone{ref: ^ref}} ->
+        MOQX.Catalog.decode(IO.iodata_to_binary(acc))
+
+      {:moqx_request_error, %MOQX.RequestError{op: :fetch, ref: ^ref, message: reason}} ->
+        {:error, reason}
+
+      {:moqx_transport_error, %MOQX.TransportError{op: :fetch, ref: ^ref, message: reason}} ->
+        {:error, reason}
+    after
+      timeout -> {:error, "timeout"}
+    end
+  end
+
+  defp normalize_namespace!(namespace) when is_binary(namespace), do: namespace
+
+  defp normalize_namespace!(namespace) do
+    raise ArgumentError,
+          "expected catalog :namespace to be a string, got: #{inspect(namespace)}"
+  end
+end

native/moqx_native/src/lib.rs

@@ -1404,7 +1404,7 @@ async fn handle_subgroup_stream(
     // just before the control loop processes the matching SubscribeOk(track_alias).
     // Wait briefly for local activation before deciding this alias is unknown.
     let has_subscription =
-        wait_for_active_subscription(inner, track_alias, Duration::from_millis(500), Duration::from_millis(5)).await;
+        wait_for_active_subscription(inner, track_alias, Duration::from_millis(2_000), Duration::from_millis(5)).await;
 
     if !has_subscription {
         return Ok(());