Apply reviewers suggestions

Author: Noarkhh

examples.livemd

@@ -11,7 +11,7 @@ System.put_env("PATH", "/opt/homebrew/bin:#{System.get_env("PATH")}")
 # Examples that don't mention them should still work.
 
 # MIX_INSTALL_CONFIG_BEGIN
-boombox = {:boombox, github: "membraneframework/boombox"}
+boombox = {:boombox, github: "membraneframework/boombox", branch: "refactor-elixir-endpoints"}
 
 # This livebook uses boombox from the master branch. If any examples happen to not work, the latest stable version of this livebook
 # can be found on https://hexdocs.pm/boombox/examples.html or in the latest github release.
@@ -24,7 +24,8 @@ Mix.install([
   :exla,
   :bumblebee,
   :websockex,
-  :membrane_simple_rtsp_server
+  :membrane_simple_rtsp_server,
+  {:coerce, ">= 1.0.2"}
 ])
 
 Nx.global_default_backend(EXLA.Backend)
@@ -573,19 +574,8 @@ reader2 =
 
 writer = Boombox.run(input: {:writer, video: :image, audio: false}, output: output)
 
-Stream.unfold(%{}, fn _state ->
+Stream.repeatedly(fn ->
   case {Boombox.read(reader1), Boombox.read(reader2)} do
-    {:finished, :finished} ->
-      nil
-
-    {{:ok, _packet}, :finished} ->
-      Boombox.close(reader1)
-      nil
-
-    {:finished, {:ok, _packet}} ->
-      Boombox.close(reader2)
-      nil
-
     {{:ok, packet1}, {:ok, packet2}} ->
       joined_image =
         Vix.Vips.Operation.join!(packet1.payload, packet2.payload, :VIPS_DIRECTION_HORIZONTAL)
@@ -598,12 +588,15 @@ Stream.unfold(%{}, fn _state ->
 
       Boombox.write(writer, packet)
 
-      {nil, %{}}
+    _finished ->
+      :eos
   end
 end)
-|> Stream.run()
+|> Enum.find(& &1 == :eos)
 
 Boombox.close(writer)
+Boombox.close(reader1)
+Boombox.close(reader2)
 ```
 
 The second cell uses `:message` endpoints, meaning that the server communicates with boomboxes by
@@ -620,39 +613,39 @@ defmodule MyServer do
 
   @impl true
   def init(args) do
-    bb1 = Boombox.run(input: args.input1, output: {:message, video: :image, audio: false})
-    bb2 = Boombox.run(input: args.input2, output: {:message, video: :image, audio: false})
+    boombox1 = Boombox.run(input: args.input1, output: {:message, video: :image, audio: false})
+    boombox2 = Boombox.run(input: args.input2, output: {:message, video: :image, audio: false})
     output_writer =
       Boombox.run(input: {:writer, video: :image, audio: false}, output: args.output)
 
     {:ok,
      %{
-       bb_states: %{
-         bb1: %{last_packet: nil, eos: false},
-         bb2: %{last_packet: nil, eos: false}
+       boombox_states: %{
+         boombox1: %{last_packet: nil, eos: false},
+         boombox2: %{last_packet: nil, eos: false}
        },
-       bbs: %{bb1 => :bb1, bb2 => :bb2},
+       boomboxes: %{boombox1 => :boombox1, boombox2 => :boombox2},
        output_writer: output_writer
      }}
   end
 
   @impl true
   def handle_info({:boombox_packet, bb, %Boombox.Packet{} = packet}, state) do
-    boombox_id = state.bbs[bb]
-    state = put_in(state.bb_states[boombox_id].last_packet, packet)
+    boombox_id = state.boomboxes[bb]
+    state = put_in(state.boombox_states[boombox_id].last_packet, packet)
 
-    if Enum.all?(Map.values(state.bb_states), &(&1.last_packet != nil)) do
+    if Enum.all?(Map.values(state.boombox_states), &(&1.last_packet != nil)) do
       joined_image =
         Vix.Vips.Operation.join!(
-          state.bb_states.bb1.last_packet.payload,
-          state.bb_states.bb2.last_packet.payload,
+          state.boombox_states.boombox1.last_packet.payload,
+          state.boombox_states.boombox2.last_packet.payload,
           :VIPS_DIRECTION_HORIZONTAL
         )
 
       pts =
         max(
-          state.bb_states.bb1.last_packet.pts,
-          state.bb_states.bb2.last_packet.pts
+          state.boombox_states.boombox1.last_packet.pts,
+          state.boombox_states.boombox2.last_packet.pts
         )
 
       packet = %Boombox.Packet{packet | payload: joined_image, pts: pts}
@@ -665,10 +658,10 @@ defmodule MyServer do
 
   @impl true
   def handle_info({:boombox_finished, bb}, state) do
-    boombox_id = state.bbs[bb]
-    state = put_in(state.bb_states[boombox_id].eos, true)
+    boombox_id = state.boomboxes[bb]
+    state = put_in(state.boombox_states[boombox_id].eos, true)
 
-    if Enum.all?(Map.values(state.bb_states), & &1.eos) do
+    if Enum.all?(Map.values(state.boombox_states), & &1.eos) do
       Boombox.close(state.output_writer)
       {:stop, :normal, state}
     else
@@ -686,7 +679,6 @@ monitor = Process.monitor(server)
 
 receive do
   {:DOWN, ^monitor, :process, ^server, reason} ->
-    IO.inspect(reason)
     :ok
 end
 ```

lib/boombox.ex

@@ -9,9 +9,9 @@ defmodule Boombox do
   require Membrane.Time
   require Membrane.Transcoder.{Audio, Video}
 
+  alias Boombox.Pipeline
   alias Membrane.HTTPAdaptiveStream
   alias Membrane.RTP
-  alias Boombox.Pipeline
 
   @elixir_endpoints [:stream, :message, :writer, :reader]
 
@@ -390,7 +390,7 @@ defmodule Boombox do
   any more packets with `write/2` and should terminate accordingly.
 
   """
-  @spec close(Writer.t() | Reader.t()) :: :ok | {:error, :incompatible_mode}
+  @spec close(Writer.t() | Reader.t()) :: :ok | {:error, :incompatible_mode | :already_finished}
   def close(%Writer{} = writer) do
     Boombox.Server.finish_consuming(writer.server_reference)
   end

lib/boombox/bin.ex

@@ -163,7 +163,8 @@ defmodule Boombox.Bin do
     spec =
       child(:boombox, %Boombox.InternalBin{
         input: opts.input || :membrane_pad,
-        output: opts.output || :membrane_pad
+        output: opts.output || :membrane_pad,
+        parent: self()
       })
 
     {[spec: spec], Map.from_struct(opts)}

lib/boombox/internal_bin.ex

@@ -135,7 +135,8 @@ defmodule Boombox.InternalBin do
     ]
 
   def_options input: [spec: input()],
-              output: [spec: output()]
+              output: [spec: output()],
+              parent: [spec: pid()]
 
   @impl true
   def handle_init(ctx, opts) do
@@ -147,8 +148,8 @@ defmodule Boombox.InternalBin do
 
     state =
       %State{
-        input: parse_endpoint_opt!(:input, opts.input),
-        output: parse_endpoint_opt!(:output, opts.output),
+        input: parse_endpoint_opt!(:input, opts.input, opts.parent),
+        output: parse_endpoint_opt!(:output, opts.output, opts.parent),
         status: :init
       }
 
@@ -730,14 +731,14 @@ defmodule Boombox.InternalBin do
     Process.sleep(500)
   end
 
-  @spec parse_endpoint_opt!(:input, input()) :: input()
-  @spec parse_endpoint_opt!(:output, output()) :: output()
-  defp parse_endpoint_opt!(direction, value) when is_binary(value) do
-    parse_endpoint_opt!(direction, {value, []})
+  @spec parse_endpoint_opt!(:input, input(), pid()) :: input()
+  @spec parse_endpoint_opt!(:output, output(), pid()) :: output()
+  defp parse_endpoint_opt!(direction, value, parent) when is_binary(value) do
+    parse_endpoint_opt!(direction, {value, []}, parent)
   end
 
   # credo:disable-for-next-line Credo.Check.Refactor.CyclomaticComplexity
-  defp parse_endpoint_opt!(direction, {value, opts}) when is_binary(value) do
+  defp parse_endpoint_opt!(direction, {value, opts}, parent) when is_binary(value) do
     uri = URI.parse(value)
     scheme = uri.scheme
     extension = if uri.path, do: Path.extname(uri.path)
@@ -769,16 +770,16 @@ defmodule Boombox.InternalBin do
       _other ->
         raise ArgumentError, "Unsupported URI: #{value} for direction: #{direction}"
     end
-    |> then(&parse_endpoint_opt!(direction, &1))
+    |> then(&parse_endpoint_opt!(direction, &1, parent))
   end
 
   # credo:disable-for-next-line Credo.Check.Refactor.CyclomaticComplexity
-  defp parse_endpoint_opt!(direction, value) when is_tuple(value) or is_atom(value) do
+  defp parse_endpoint_opt!(direction, value, parent) when is_tuple(value) or is_atom(value) do
     case value do
       {endpoint_type, location}
       when is_binary(location) and direction == :input and
              StorageEndpoints.is_storage_endpoint_type(endpoint_type) ->
-        parse_endpoint_opt!(:input, {endpoint_type, location, []})
+        parse_endpoint_opt!(:input, {endpoint_type, location, []}, parent)
 
       {endpoint_type, location, opts}
       when endpoint_type in [:h264, :h265] and is_binary(location) and direction == :input ->
@@ -819,7 +820,7 @@ defmodule Boombox.InternalBin do
         value
 
       {:whip, uri} when is_binary(uri) ->
-        parse_endpoint_opt!(direction, {:whip, uri, []})
+        parse_endpoint_opt!(direction, {:whip, uri, []}, parent)
 
       {:whip, uri, opts} when is_binary(uri) and is_list(opts) and direction == :input ->
         if Keyword.keyword?(opts), do: {:webrtc, value}
@@ -850,9 +851,9 @@ defmodule Boombox.InternalBin do
       {:rtp, opts} ->
         if Keyword.keyword?(opts), do: value
 
-      {elixir_endpoint, process, opts}
-      when is_pid(process) and elixir_endpoint in @elixir_endpoint_types ->
-        if Keyword.keyword?(opts), do: value
+      {elixir_endpoint, opts}
+      when elixir_endpoint in @elixir_endpoint_types ->
+        if Keyword.keyword?(opts), do: {elixir_endpoint, parent, opts}
 
       {:srt, server_awaiting_accept}
       when direction == :input and is_pid(server_awaiting_accept) ->