From 2872b8ff2fce537c9249f3125d3d2004e3da6110 Mon Sep 17 00:00:00 2001
From: Yordis Prieto <yordis.prieto@gmail.com>
Date: Tue, 27 Feb 2024 18:11:19 -0500
Subject: [PATCH] feat: add pause consumer

---
 lib/gnat/jetstream/api/consumer.ex | 28 ++++++++++++++++++++++++++++
 1 file changed, 28 insertions(+)

diff --git a/lib/gnat/jetstream/api/consumer.ex b/lib/gnat/jetstream/api/consumer.ex
index e24002d..80581cb 100644
--- a/lib/gnat/jetstream/api/consumer.ex
+++ b/lib/gnat/jetstream/api/consumer.ex
@@ -275,6 +275,34 @@ defmodule Gnat.Jetstream.API.Consumer do
     end
   end
 
+  @doc """
+  Pause a consumer until a specific time.
+
+  ## Examples
+
+      iex> {:ok, _response} = Gnat.Jetstream.API.Stream.create(:gnat, %Gnat.Jetstream.API.Stream{name: "astream", subjects: ["subject"]})
+      iex> {:ok, _response} = Gnat.Jetstream.API.Consumer.create(:gnat, %Gnat.Jetstream.API.Consumer{durable_name: "consumer", stream_name: "astream"})
+      iex> pause_until = DateTime.add(DateTime.utc_now(), 1, :hour)
+      iex> Gnat.Jetstream.API.Consumer.pause(:gnat, "astream", "consumer", pause_until)
+      :ok
+
+  """
+  @spec pause(
+    conn :: Gnat.t(),
+    stream_name :: binary(),
+    consumer_name :: binary(),
+    pause_until :: DateTime.t(),
+    opts :: [domain: nil | binary()]
+  ) ::
+    :ok | {:error, any()}
+def pause(conn, stream_name, consumer_name, pause_until, opts \\ []) do
+  topic = "#{js_api(opts[:domain])}.CONSUMER.PAUSE.#{stream_name}.#{consumer_name}"
+  payload = Jason.encode!(%{ pause_until: pause_until })
+  with {:ok, _response} <- request(conn, topic, payload) do
+    :ok
+  end
+end
+
   @doc """
   Information about the consumer.