Add MOQ Lite publisher operations

Author: dmorn

.scratch/moq-lite-04-protocol/PRD.md

@@ -1,6 +1,6 @@
 # MOQ Lite 04 protocol layer
 
-Status: in-progress
+Status: done
 
 ## Problem Statement
 
@@ -171,11 +171,16 @@ Delivered so far:
 - Role-neutral subscriber operations on `MOQX.MOQLite04` for
   AnnounceInterest, Subscribe, SubscribeUpdate, Fetch, Probe, repeated Probe,
   and Goaway transaction streams
+- Role-neutral publisher operations on `MOQX.MOQLite04` for accepting
+  peer-opened streams, sending Announce, SubscribeOk, and SubscribeDrop
+  responses, and publishing Group plus Frame messages on publisher-created
+  unidirectional streams
+- A client-level support-transport smoke flow that subscribes, accepts the
+  subscription, sends SubscribeOk, publishes a Group, and delivers the original
+  Frame payload through public client APIs
 - Session tests for Announce, Subscribe, Fetch, Probe, Goaway, Group,
   stream lifecycle, support transport normalized events, and the
   reference-style subscribe/group/frame smoke flow
-- Client/API follow-up issues:
-  - issue 08: role-neutral publisher operations and smoke flow
 
 ## References
 

.scratch/moq-lite-04-protocol/issues/08-add-role-neutral-moq-lite-publisher-operations.md

@@ -2,7 +2,7 @@
 
 # Add role-neutral MOQ Lite 04 publisher operations
 
-Status: ready-for-agent
+Status: done
 Type: enhancement
 
 ## Parent
@@ -25,26 +25,25 @@ flow that mirrors the reducer smoke flow from issue 04, but now passes through
 
 ## Acceptance criteria
 
-- [ ] Public functions exist on `MOQX.MOQLite04` for sending Announce messages
+- [x] Public functions exist on `MOQX.MOQLite04` for sending Announce messages
       on an Announce transaction stream.
-- [ ] Public functions exist for sending SubscribeOk and SubscribeDrop on a
+- [x] Public functions exist for sending SubscribeOk and SubscribeDrop on a
       Subscribe transaction stream.
-- [ ] Public functions exist for opening a publisher-created Group stream and
+- [x] Public functions exist for opening a publisher-created Group stream and
       sending Group plus Frame messages.
-- [ ] SubscribeDrop before SubscribeOk is rejected through the client API using
+- [x] SubscribeDrop before SubscribeOk is rejected through the client API using
       the Session reducer's structured error.
-- [ ] Duplicate Announce status for one suffix is rejected through the client
+- [x] Duplicate Announce status for one suffix is rejected through the client
       API using the Session reducer's structured error.
-- [ ] Group publishing is rejected unless the peer subscription is active.
-- [ ] `connect/2` remains role-neutral; no publisher/subscriber connect mode is
+- [x] Group publishing is rejected unless the peer subscription is active.
+- [x] `connect/2` remains role-neutral; no publisher/subscriber connect mode is
       added.
-- [ ] A support-transport smoke test proves one endpoint can subscribe and the
+- [x] A support-transport smoke test proves one endpoint can subscribe and the
       other can accept the subscription, send SubscribeOk, publish a Group, and
       deliver the original Frame payload.
-- [ ] The implementation preserves the pure Session reducer boundary and does
+- [x] The implementation preserves the pure Session reducer boundary and does
       not match raw `quicer` messages.
 
 ## Blocked by
 
 - `.scratch/moq-lite-04-protocol/issues/07-add-role-neutral-moq-lite-subscriber-operations.md`
-

lib/moqx/moq_lite_04.ex

@@ -74,6 +74,14 @@ defmodule MOQX.MOQLite04 do
           | {:timeout, Client.t()}
   def recv(client, timeout \\ :infinity), do: Client.recv(client, timeout)
 
+  @doc """
+  Accepts a peer-opened transport stream and starts active delivery.
+  """
+  @spec accept_stream(Client.t(), timeout()) ::
+          {:ok, Client.t(), MOQX.Transport.Stream.t()}
+          | {:error, Client.t(), term()}
+  def accept_stream(client, timeout \\ :infinity), do: Client.accept_stream(client, timeout)
+
   @doc """
   Opens an Announce transaction stream and sends `ANNOUNCE_INTEREST`.
   """
@@ -131,6 +139,38 @@ defmodule MOQX.MOQLite04 do
           | {:error, Client.t(), term(), [term()]}
   def goaway(client, message), do: Client.goaway(client, message)
 
+  @doc """
+  Sends `ANNOUNCE` on an existing Announce transaction stream.
+  """
+  @spec announce(Client.t(), MOQX.Transport.Stream.t(), Announce.t()) ::
+          {:ok, Client.t(), MOQX.Transport.Stream.t(), [term()]}
+          | {:error, Client.t(), term(), [term()]}
+  def announce(client, stream, message), do: Client.announce(client, stream, message)
+
+  @doc """
+  Sends `SUBSCRIBE_OK` on an existing Subscribe transaction stream.
+  """
+  @spec subscribe_ok(Client.t(), MOQX.Transport.Stream.t(), SubscribeOk.t()) ::
+          {:ok, Client.t(), MOQX.Transport.Stream.t(), [term()]}
+          | {:error, Client.t(), term(), [term()]}
+  def subscribe_ok(client, stream, message), do: Client.subscribe_ok(client, stream, message)
+
+  @doc """
+  Sends `SUBSCRIBE_DROP` on an existing Subscribe transaction stream.
+  """
+  @spec subscribe_drop(Client.t(), MOQX.Transport.Stream.t(), SubscribeDrop.t()) ::
+          {:ok, Client.t(), MOQX.Transport.Stream.t(), [term()]}
+          | {:error, Client.t(), term(), [term()]}
+  def subscribe_drop(client, stream, message), do: Client.subscribe_drop(client, stream, message)
+
+  @doc """
+  Opens a publisher Group stream and sends `GROUP` followed by `FRAME` messages.
+  """
+  @spec publish_group(Client.t(), Group.t(), [Frame.t()]) ::
+          {:ok, Client.t(), MOQX.Transport.Stream.t(), [term()]}
+          | {:error, Client.t(), term(), [term()]}
+  def publish_group(client, group, frames), do: Client.publish_group(client, group, frames)
+
   @doc """
   Returns the draft-04 numeric stream type ID for a known stream type.
   """

lib/moqx/moq_lite_04/client.ex

@@ -57,6 +57,8 @@ defmodule MOQX.MOQLite04.Client do
   @type operation_result ::
           {:ok, t(), Stream.t(), [term()]} | {:error, t(), term(), [term()]}
 
+  @type accept_result :: {:ok, t(), Stream.t()} | {:error, t(), term()}
+
   @doc """
   Connects to a native QUIC MOQ Lite draft-04 endpoint.
   """
@@ -136,6 +138,27 @@ defmodule MOQX.MOQLite04.Client do
     end
   end
 
+  @doc """
+  Accepts a peer-opened stream and starts active transport delivery.
+  """
+  @spec accept_stream(t(), timeout()) :: accept_result()
+  def accept_stream(%__MODULE__{} = client, timeout \\ :infinity) do
+    with {:ok, stream, context} <-
+           Transport.accept_stream(client.context, client.connection, [], timeout),
+         {:ok, context} <- Transport.set_active(context, stream, true) do
+      {:ok, put_context(client, context), stream}
+    else
+      {:error, reason, context} ->
+        error =
+          Error.new(:transport_action_failed,
+            action: :accept_stream,
+            details: %{transport_reason: reason}
+          )
+
+        {:error, put_context(client, context), error}
+    end
+  end
+
   @doc """
   Opens an Announce transaction stream and sends `ANNOUNCE_INTEREST`.
   """
@@ -196,6 +219,43 @@ defmodule MOQX.MOQLite04.Client do
     open_transaction(client, :goaway, message)
   end
 
+  @doc """
+  Sends `ANNOUNCE` on an existing Announce transaction stream.
+  """
+  @spec announce(t(), Stream.t(), MOQLite04.Announce.t()) :: operation_result()
+  def announce(%__MODULE__{} = client, %Stream{} = stream, %MOQLite04.Announce{} = message) do
+    send_on_transaction(client, stream, :announce, message)
+  end
+
+  @doc """
+  Sends `SUBSCRIBE_OK` on an existing Subscribe transaction stream.
+  """
+  @spec subscribe_ok(t(), Stream.t(), MOQLite04.SubscribeOk.t()) :: operation_result()
+  def subscribe_ok(%__MODULE__{} = client, %Stream{} = stream, %MOQLite04.SubscribeOk{} = message) do
+    send_on_transaction(client, stream, :subscribe, message)
+  end
+
+  @doc """
+  Sends `SUBSCRIBE_DROP` on an existing Subscribe transaction stream.
+  """
+  @spec subscribe_drop(t(), Stream.t(), MOQLite04.SubscribeDrop.t()) :: operation_result()
+  def subscribe_drop(
+        %__MODULE__{} = client,
+        %Stream{} = stream,
+        %MOQLite04.SubscribeDrop{} = message
+      ) do
+    send_on_transaction(client, stream, :subscribe, message)
+  end
+
+  @doc """
+  Opens a publisher Group stream and sends `GROUP` followed by `FRAME` messages.
+  """
+  @spec publish_group(t(), MOQLite04.Group.t(), [MOQLite04.Frame.t()]) :: operation_result()
+  def publish_group(%__MODULE__{} = client, %MOQLite04.Group{} = group, frames)
+      when is_list(frames) do
+    open_group_stream(client, [group | frames])
+  end
+
   defp reject_mode(opts) do
     if Keyword.has_key?(opts, :mode) do
       {:error, {:unsupported_option, :mode}}
@@ -279,8 +339,30 @@ defmodule MOQX.MOQLite04.Client do
     end
   end
 
+  defp open_group_stream(%__MODULE__{} = client, messages) do
+    case Transport.open_stream(client.context, client.connection, direction: :unidirectional) do
+      {:ok, stream, context} ->
+        client
+        |> put_context(context)
+        |> send_on_stream(stream, :group, messages)
+
+      {:error, reason, context} ->
+        error =
+          Error.new(:transport_action_failed,
+            action: {:open_stream, :group},
+            details: %{transport_reason: reason}
+          )
+
+        {:error, put_context(client, context), error, []}
+    end
+  end
+
   defp send_on_transaction(%__MODULE__{} = client, %Stream{} = stream, stream_type, message) do
-    case command(client, {:send, stream, stream_type, [message]}) do
+    send_on_stream(client, stream, stream_type, [message])
+  end
+
+  defp send_on_stream(%__MODULE__{} = client, %Stream{} = stream, stream_type, messages) do
+    case command(client, {:send, stream, stream_type, messages}) do
       {:ok, client, events} -> {:ok, client, stream, events}
       {:error, client, reason, events} -> {:error, client, reason, events}
     end