Page over stream

lib/gnat.ex

@@ -25,14 +25,18 @@ defmodule Gnat do
   * `sid` - The subscription ID corresponding to this message. You generally won't need to use this value directly.
   * `reply_to` - A topic supplied for expected replies
   * `headers` - A set of NATS message headers on the message
+  * `status` - Similar to an HTTP status, this is present for messages with headers and can indicate the specific purpose of a message. Example `status: "408"`
+  * `description` - A string description of the `status`
   """
   @type message :: %{
     required(:gnat) => t(),
     required(:topic) => binary(),
     required(:body) => iodata(),
     required(:sid) => non_neg_integer(),
     optional(:reply_to) => binary(),
-    optional(:headers) => headers()
+    optional(:headers) => headers(),
+    optional(:status) => String.t(),
+    optional(:description) => String.t()
   }
   @type sent_message :: {:msg, message()}
 

lib/gnat/jetstream/jetstream.ex

@@ -10,7 +10,9 @@ Sends `AckAck` acknowledgement to the server.
 Acknowledges a message was completely handled.
 """
 @spec ack(message :: Gnat.message()) :: :ok
-def ack(message)
+def ack(%{reply_to: nil}) do
+  {:error, "Cannot ack message with no reply-to"}
+end
 
 def ack(%{gnat: gnat, reply_to: reply_to}) do
   Gnat.pub(gnat, reply_to, "")
@@ -23,7 +25,9 @@ Acknowledges the message was handled and requests delivery of the next message t
 subject. Only applies to Pull-mode.
 """
 @spec ack_next(message :: Gnat.message(), consumer_subject :: binary()) :: :ok
-def ack_next(message, consumer_subject)
+def ack_next(%{reply_to: nil}, _consumer_subject) do
+  {:error, "Cannot ack message with no reply-to"}
+end
 
 def ack_next(%{gnat: gnat, reply_to: reply_to}, consumer_subject) do
   Gnat.pub(gnat, reply_to, "+NXT", reply_to: consumer_subject)
@@ -36,7 +40,9 @@ Signals that the message will not be processed now and processing can move onto
 NAK'd message will be retried.
 """
 @spec nack(message :: Gnat.message()) :: :ok
-def nack(message)
+def nack(%{reply_to: nil}) do
+  {:error, "Cannot ack message with no reply-to"}
+end
 
 def nack(%{gnat: gnat, reply_to: reply_to}) do
   Gnat.pub(gnat, reply_to, "-NAK")
@@ -49,7 +55,9 @@ When sent before the `AckWait` period indicates that work is ongoing and the per
 extended by another equal to `AckWait`.
 """
 @spec ack_progress(message :: Gnat.message()) :: :ok
-def ack_progress(message)
+def ack_progress(%{reply_to: nil}) do
+  {:error, "Cannot ack message with no reply-to"}
+end
 
 def ack_progress(%{gnat: gnat, reply_to: reply_to}) do
   Gnat.pub(gnat, reply_to, "+WPI")
@@ -62,7 +70,9 @@ Instructs the server to stop redelivery of a message without acknowledging it as
 processed.
 """
 @spec ack_term(message :: Gnat.message()) :: :ok
-def ack_term(message)
+def ack_term(%{reply_to: nil}) do
+  {:error, "Cannot ack message with no reply-to"}
+end
 
 def ack_term(%{gnat: gnat, reply_to: reply_to}) do
   Gnat.pub(gnat, reply_to, "+TERM")

lib/gnat/jetstream/pager.ex

@@ -0,0 +1,75 @@
+defmodule Gnat.Jetstream.Pager do
+  @moduledoc false
+
+  alias Gnat.Jetstream
+  alias Gnat.Jetstream.API.{Consumer, Util}
+
+  @opaque pager :: map()
+  @type message :: Gnat.message()
+
+  def init(conn, stream_name, opts) do
+    name = "gnat_stream_pager_#{Util.nuid()}"
+
+    first_seq = Keyword.fetch!(opts, :from_seq)
+
+    consumer = %Consumer{
+      stream_name: stream_name,
+      durable_name: name,
+      ack_policy: :explicit,
+      ack_wait: 30_000_000_000,
+      deliver_policy: :by_start_sequence,
+      description: "Gnat Stream Pager",
+      opt_start_seq: first_seq,
+      replay_policy: :instant,
+      inactive_threshold: 30_000_000_000
+    }
+    inbox = Util.reply_inbox()
+
+    with {:ok, _config} <- Consumer.create(conn, consumer),
+         {:ok, sub} <- Gnat.sub(conn, self(), inbox) do
+      state =
+        %{
+          conn: conn,
+          stream_name: stream_name,
+          consumer_name: name,
+          domain: nil,
+          inbox: inbox,
+          batch: 10,
+          sub: sub
+        }
+
+      {:ok, state}
+    end
+  end
+
+  @spec page(pager()) :: {:page, list(message())} | {:done, list(message())} | {:error, term()}
+  def page(%{conn: conn, batch: batch} = state) do
+    opts = [batch: batch, no_wait: true]
+    with :ok <- Consumer.request_next_message(conn, state.stream_name, state.consumer_name, state.inbox, state.domain, opts) do
+      receive_messages(state, [])
+    end
+  end
+
+  def cleanup(%{conn: conn} = state) do
+    with :ok <- Gnat.unsub(conn, state.sub),
+         :ok <- Consumer.delete(conn, state.stream_name, state.consumer_name, state.domain) do
+      :ok
+    end
+  end
+
+  defp receive_messages(%{batch: batch}, messages) when length(messages) == batch do
+    {:page, Enum.reverse(messages)}
+  end
+
+  defp receive_messages(%{sub: sid} = state, messages) do
+    receive do
+      {:msg, %{sid: ^sid, status: "408"}} ->
+        {:done, Enum.reverse(messages)}
+
+      {:msg, %{sid: ^sid} = message} ->
+        with :ok <- Jetstream.ack(message) do
+          receive_messages(state, [message | messages ])
+        end
+    end
+  end
+end