Add public function for closing boombox reader

Author: Noarkhh

lib/boombox.ex

@@ -371,16 +371,24 @@ defmodule Boombox do
   end
 
   @doc """
-  Informs Boombox that it will not be provided any more packets with `write/2` and should terminate
-  accordingly.
+  Gracefully terminates Boombox when using `:reader` or `:writer` endpoints.
+
+  When using `:reader` endpoint on output informs Boombox that no more packets will be read
+  from it with `read/1` and that it should terminate accordingly.
+
+  When using `:writer` endpoint on input informs Boombox that it will not be provided
+  any more packets with `write/2` and should terminate accordingly.
 
-  Can be called only when using `:writer` endpoint on input.
   """
-  @spec close(Writer.t()) :: :finished | {:error, :incompatible_mode}
-  def close(writer) do
+  @spec close(Writer.t() | Reader.t()) :: :finished | {:error, :incompatible_mode}
+  def close(%Writer{} = writer) do
     Boombox.Server.finish_consuming(writer.server_reference)
   end
 
+  def close(%Reader{} = reader) do
+    Boombox.Server.finish_producing(reader.server_reference)
+  end
+
   @endpoint_opts [:input, :output]
   defp validate_opts!(stream, opts) do
     opts = opts |> Keyword.validate!(@endpoint_opts) |> Map.new()

lib/boombox/server.ex

@@ -175,6 +175,17 @@ defmodule Boombox.Server do
     GenServer.call(server, :produce_packet)
   end
 
+  @doc """
+  Informs Boombox that no more packets will be read and shouldn't be produced.
+  Can be called only when Boombox is in `:producing` mode.
+  """
+  @spec finish_producing(t()) ::
+          {:finished, serialized_boombox_packet() | Boombox.Packet.t()}
+          | {:error, :incompatible_mode}
+  def finish_producing(server) do
+    GenServer.call(server, :finish_producing)
+  end
+
   @impl true
   def init(opts) do
     {:ok,
@@ -390,6 +401,25 @@ defmodule Boombox.Server do
     {{:error, :incompatible_mode}, state}
   end
 
+  @spec handle_request(:finish_producing, State.t()) ::
+          {{:finished, serialized_boombox_packet() | Boombox.Packet.t()}
+           | {:error, :incompatible_mode}, State.t()}
+  defp handle_request(
+         :finish_producing,
+         %State{boombox_mode: :producing, boombox_pid: boombox_pid} = state
+       ) do
+    send(boombox_pid, :finish_producing)
+
+    receive do
+      {:finished, packet, ^boombox_pid} ->
+        {{:finished, packet}, state}
+    end
+  end
+
+  defp handle_request(:finish_producing, %State{boombox_mode: _other_mode} = state) do
+    {{:error, :incompatible_mode}, state}
+  end
+
   @spec handle_request(term(), State.t()) :: {{:error, :invalid_request}, State.t()}
   defp handle_request(_invalid_request, state) do
     {{:error, :invalid_request}, state}