Document the Pager module and add support for from_datetime (#216)

* Document the Pager module and add support for from_datetime

* mix format

Author: mmmries

lib/gnat/jetstream/pager.ex

@@ -1,12 +1,37 @@
 defmodule Gnat.Jetstream.Pager do
-  @moduledoc false
+  @moduledoc """
+  Page through all the messages in a stream
+
+  This module provides a synchronous API to inspect the messages in a stream.
+  You can use the reduce module to write a simple function that works like `Enum.reduce` across each message individually.
+  If you want to handle messages in batches, you can use the `init` + `page` functions to accomplish that.
+  """
 
   alias Gnat.Jetstream
   alias Gnat.Jetstream.API.{Consumer, Util}
 
   @opaque pager :: map()
   @type message :: Gnat.message()
-
+  @type opts :: list(opt())
+
+  @typedoc """
+  Options you can pass to the pager
+
+  * `batch` controls the maximum number of messages we'll pull in each page/batch (default 10)
+  * `domain` You can specify a jetstream domain if needed
+  * `from_datetime` Only page through messages recorded on or after this datetime
+  * `from_seq` Only page through messages with a sequence number equal or above this option
+  * `headers_only` You can pass `true` to this if you only want to see the headers from each message. Can be useful to get metadata without having to receieve large body payloads.
+
+  """
+  @type opt ::
+          {:batch, non_neg_integer()}
+          | {:domain, String.t()}
+          | {:from_datetime, DateTime.t()}
+          | {:from_seq, non_neg_integer}
+          | {:headers_only, boolean()}
+
+  @spec init(Gnat.t(), String.t(), opts()) :: {:ok, pager()} | {:error, term()}
   def init(conn, stream_name, opts) do
     domain = Keyword.get(opts, :domainl)
 
@@ -55,6 +80,26 @@ defmodule Gnat.Jetstream.Pager do
     end
   end
 
+  @doc """
+  Similar to Enum.reduce but you can iterate through all messages in a stream
+
+  ```
+  # Assume we have a stream with messages like "1", "2", ... "10"
+  Gnat.Jetstream.Pager.reduce(:gnat, "NUMBERS_STREAM", [batch_size: 5], 0, fn(message, total) ->
+    num = String.to_integer(message.body)
+    total + num
+  end)
+
+  # => {:ok, 55}
+  ```
+  """
+  @spec reduce(
+          Gnat.t(),
+          String.t(),
+          opts(),
+          Enum.acc(),
+          (Gnat.message(), Enum.acc() -> Enum.acc())
+        ) :: {:ok, Enum.acc()} | {:error, term()}
   def reduce(conn, stream_name, opts, initial_state, fun) do
     with {:ok, pager} <- init(conn, stream_name, opts) do
       page_through(pager, initial_state, fun)
@@ -109,12 +154,17 @@ defmodule Gnat.Jetstream.Pager do
 
   ## Helpers for accepting user options
   defp apply_opts_to_consumer(consumer = %Consumer{}, opts) do
-    case Keyword.get(opts, :from_seq) do
-      nil ->
+    from = {Keyword.get(opts, :from_seq), Keyword.get(opts, :from_datetime)}
+
+    case from do
+      {nil, nil} ->
         consumer
 
-      seq when is_integer(seq) ->
+      {seq, _} when is_integer(seq) ->
         %Consumer{consumer | deliver_policy: :by_start_sequence, opt_start_seq: seq}
+
+      {_, %DateTime{} = dt} ->
+        %Consumer{consumer | deliver_policy: :by_start_time, opt_start_time: dt}
     end
   end
 end

test/jetstream/pager_test.exs

@@ -29,6 +29,31 @@ defmodule Gnat.Jetstream.PagerTest do
     Stream.delete(:gnat, "pager_a")
   end
 
+  test "paging from a datetime" do
+    {:ok, _stream} = create_stream("pager_b")
+
+    Enum.each(1..50, fn i ->
+      :ok = Gnat.pub(:gnat, "input.pager_b", "#{i}")
+    end)
+
+    timestamp = DateTime.utc_now()
+
+    Enum.each(51..100, fn i ->
+      :ok = Gnat.pub(:gnat, "input.pager_b", "#{i}")
+    end)
+
+    {:ok, res} =
+      Pager.reduce(:gnat, "pager_b", [from_datetime: timestamp], 0, fn msg, total ->
+        total + String.to_integer(msg.body)
+      end)
+
+    # The datetime isn't exactly synced between nats and the client
+    # so we use a pretty fuzzy check here. This test is mostly to
+    # provide coverage for accepting the option
+    assert res <= 5050
+    assert res >= 3775
+  end
+
   defp create_stream(name) do
     stream = %Stream{
       name: name,