Fix naming

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", branch: "refactor-elixir-endpoints"}
+boombox = {:boombox, github: "membraneframework/boombox"}
 
 # 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.
@@ -575,27 +575,27 @@ writer = Boombox.run(input: {:writer, video: :image, audio: false}, output: outp
 
 Stream.unfold(%{}, fn _state ->
   case {Boombox.read(reader1), Boombox.read(reader2)} do
-    {:finished, :finished} -> 
+    {:finished, :finished} ->
       nil
 
-    {{:ok, _packet}, :finished} -> 
+    {{:ok, _packet}, :finished} ->
       Boombox.close(reader1)
       nil
-      
-    {:finished, {:ok, _packet}} -> 
+
+    {:finished, {:ok, _packet}} ->
       Boombox.close(reader2)
       nil
-      
-    {{:ok, packet1}, {:ok, packet2}} -> 
+
+    {{:ok, packet1}, {:ok, packet2}} ->
       joined_image =
         Vix.Vips.Operation.join!(packet1.payload, packet2.payload, :VIPS_DIRECTION_HORIZONTAL)
-    
+
       packet = %Boombox.Packet{
         pts: max(packet1.pts, packet2.pts),
         payload: joined_image,
         kind: :video
       }
-    
+
       Boombox.write(writer, packet)
 
       {nil, %{}}
@@ -649,7 +649,7 @@ defmodule MyServer do
           :VIPS_DIRECTION_HORIZONTAL
         )
 
-      pts = 
+      pts =
         max(
           state.bb_states.bb1.last_packet.pts,
           state.bb_states.bb2.last_packet.pts

lib/boombox.ex

@@ -211,13 +211,13 @@ defmodule Boombox do
 
     case opts do
       %{input: {:stream, _stream_opts}} ->
-        procs = Pipeline.start_pipeline(opts)
-        source = Pipeline.await_source_ready()
+        procs = Pipeline.start(opts)
+        source = await_source_ready()
         consume_stream(stream, source, procs)
 
       %{output: {:stream, _stream_opts}} ->
-        procs = Pipeline.start_pipeline(opts)
-        sink = Pipeline.await_sink_ready()
+        procs = Pipeline.start(opts)
+        sink = await_sink_ready()
         produce_stream(sink, procs)
 
       %{input: {:writer, _writer_opts}} ->
@@ -236,8 +236,8 @@ defmodule Boombox do
 
       opts ->
         opts
-        |> Pipeline.start_pipeline()
-        |> Pipeline.await_pipeline()
+        |> Pipeline.start()
+        |> await_pipeline()
     end
   end
 
@@ -276,17 +276,17 @@ defmodule Boombox do
 
     case opts do
       %{input: {:stream, _stream_opts}} ->
-        procs = Pipeline.start_pipeline(opts)
-        source = Pipeline.await_source_ready()
+        procs = Pipeline.start(opts)
+        source = await_source_ready()
 
         Task.async(fn ->
           Process.monitor(procs.supervisor)
           consume_stream(stream, source, procs)
         end)
 
       %{output: {:stream, _stream_opts}} ->
-        procs = Pipeline.start_pipeline(opts)
-        sink = Pipeline.await_sink_ready()
+        procs = Pipeline.start(opts)
+        sink = await_sink_ready()
         produce_stream(sink, procs)
 
       %{input: {:writer, _writer_opts}} ->
@@ -306,23 +306,23 @@ defmodule Boombox do
       # In case of rtmp, rtmps, rtp, rtsp, we need to wait for the tcp/udp server to be ready
       # before returning from async/2.
       %{input: {protocol, _opts}} when protocol in [:rtmp, :rtmps, :rtp, :rtsp, :srt] ->
-        procs = Pipeline.start_pipeline(opts)
+        procs = Pipeline.start(opts)
 
         task =
           Task.async(fn ->
             Process.monitor(procs.supervisor)
-            Pipeline.await_pipeline(procs)
+            await_pipeline(procs)
           end)
 
         await_external_resource_ready()
         task
 
       opts ->
-        procs = Pipeline.start_pipeline(opts)
+        procs = Pipeline.start(opts)
 
         Task.async(fn ->
           Process.monitor(procs.supervisor)
-          Pipeline.await_pipeline(procs)
+          await_pipeline(procs)
         end)
     end
   end
@@ -471,7 +471,7 @@ defmodule Boombox do
 
       _state ->
         send(source, {:boombox_eos, self()})
-        Pipeline.await_pipeline(procs)
+        await_pipeline(procs)
     end
   end
 
@@ -495,12 +495,39 @@ defmodule Boombox do
         end
       end,
       fn
-        %{procs: procs} -> Pipeline.terminate_pipeline(procs)
+        %{procs: procs} -> terminate_pipeline(procs)
         :eos -> :ok
       end
     )
   end
 
+  @spec terminate_pipeline(Pipeline.procs()) :: :ok
+  defp terminate_pipeline(procs) do
+    Membrane.Pipeline.terminate(procs.pipeline)
+    await_pipeline(procs)
+  end
+
+  @spec await_pipeline(Pipeline.procs()) :: :ok
+  defp await_pipeline(%{supervisor: supervisor}) do
+    receive do
+      {:DOWN, _monitor, :process, ^supervisor, _reason} -> :ok
+    end
+  end
+
+  @spec await_source_ready() :: pid()
+  defp await_source_ready() do
+    receive do
+      {:boombox_elixir_source, source} -> source
+    end
+  end
+
+  @spec await_sink_ready() :: pid()
+  defp await_sink_ready() do
+    receive do
+      {:boombox_elixir_sink, sink} -> sink
+    end
+  end
+
   # Waits for the external resource to be ready.
   # This is used to wait for the tcp/udp server to be ready before returning from async/2.
   # It is used for rtmp, rtmps, rtp, rtsp.

lib/boombox/internal_bin.ex

@@ -336,7 +336,7 @@ defmodule Boombox.InternalBin do
   end
 
   @impl true
-  def handle_element_end_of_stream(:elixir_stream_sink, Pad.ref(:input, id), _ctx, state) do
+  def handle_element_end_of_stream(:elixir_sink, Pad.ref(:input, id), _ctx, state) do
     eos_info = List.delete(state.eos_info, id)
     state = %{state | eos_info: eos_info}
 

lib/boombox/internal_bin/elixir_endpoints.ex

@@ -26,14 +26,14 @@ defmodule Boombox.InternalBin.ElixirEndpoints do
       |> Map.new(fn
         :video ->
           {:video,
-           get_child(:elixir_stream_source)
+           get_child(:elixir_source)
            |> via_out(Pad.ref(:output, :video))
            |> child(%SWScale.Converter{format: :I420})
            |> child(%Membrane.H264.FFmpeg.Encoder{profile: :baseline, preset: :ultrafast})}
 
         :audio ->
           {:audio,
-           get_child(:elixir_stream_source)
+           get_child(:elixir_source)
            |> via_out(Pad.ref(:output, :audio))}
       end)
 
@@ -43,7 +43,7 @@ defmodule Boombox.InternalBin.ElixirEndpoints do
         :pull -> %PullSource{producer: producer}
       end
 
-    spec_builder = child(:elixir_stream_source, source_definition)
+    spec_builder = child(:elixir_source, source_definition)
 
     %Ready{track_builders: builders, spec_builder: spec_builder}
   end
@@ -79,31 +79,31 @@ defmodule Boombox.InternalBin.ElixirEndpoints do
     spec =
       [
         spec_builder,
-        child(:elixir_stream_sink, sink_definition),
+        child(:elixir_sink, sink_definition),
         Enum.map(track_builders, fn
           {:audio, builder} ->
             builder
-            |> child(:elixir_stream_audio_transcoder, %Membrane.Transcoder{
+            |> child(:elixir_audio_transcoder, %Membrane.Transcoder{
               output_stream_format: Membrane.RawAudio
             })
             |> maybe_plug_resampler(options)
             |> maybe_plug_realtimer(:audio, pace_control, is_input_realtime)
             |> via_in(Pad.ref(:input, :audio))
-            |> get_child(:elixir_stream_sink)
+            |> get_child(:elixir_sink)
 
           {:video, builder} ->
             builder
-            |> child(:elixir_stream_video_transcoder, %Membrane.Transcoder{
+            |> child(:elixir_video_transcoder, %Membrane.Transcoder{
               output_stream_format: Membrane.RawVideo
             })
-            |> child(:elixir_stream_rgb_converter, %SWScale.Converter{
+            |> child(:elixir_rgb_converter, %SWScale.Converter{
               format: :RGB,
               output_width: options[:video_width],
               output_height: options[:video_height]
             })
             |> maybe_plug_realtimer(:video, pace_control, is_input_realtime)
             |> via_in(Pad.ref(:input, :video))
-            |> get_child(:elixir_stream_sink)
+            |> get_child(:elixir_sink)
         end),
         Enum.map(to_ignore, fn {_track, builder} -> builder |> child(Membrane.Debug.Sink) end)
       ]
@@ -116,7 +116,7 @@ defmodule Boombox.InternalBin.ElixirEndpoints do
   defp maybe_plug_realtimer(builder, kind, true, false) do
     builder
     |> via_in(:input, toilet_capacity: @realtimer_toilet_capacity)
-    |> child({:elixir_stream, kind, :realtimer}, Membrane.Realtimer)
+    |> child({:elixir, kind, :realtimer}, Membrane.Realtimer)
   end
 
   defp maybe_plug_realtimer(builder, _kind, _pace_control, _is_input_realtime), do: builder

lib/boombox/pipeline.ex

@@ -9,12 +9,12 @@ defmodule Boombox.Pipeline do
         }
   @type procs :: %{pipeline: pid(), supervisor: pid()}
 
-  @spec start_pipeline(opts_map()) :: procs()
-  def start_pipeline(opts) do
+  @spec start(opts_map()) :: procs()
+  def start(opts) do
     opts =
       opts
-      |> Map.update!(:input, &resolve_stream_endpoint(&1, self()))
-      |> Map.update!(:output, &resolve_stream_endpoint(&1, self()))
+      |> Map.update!(:input, &resolve_elixir_endpoint(&1, self()))
+      |> Map.update!(:output, &resolve_elixir_endpoint(&1, self()))
       |> Map.put(:parent, self())
 
     {:ok, supervisor, pipeline} =
@@ -24,33 +24,6 @@ defmodule Boombox.Pipeline do
     %{supervisor: supervisor, pipeline: pipeline}
   end
 
-  @spec terminate_pipeline(procs()) :: :ok
-  def terminate_pipeline(procs) do
-    Membrane.Pipeline.terminate(procs.pipeline)
-    await_pipeline(procs)
-  end
-
-  @spec await_pipeline(procs()) :: :ok
-  def await_pipeline(%{supervisor: supervisor}) do
-    receive do
-      {:DOWN, _monitor, :process, ^supervisor, _reason} -> :ok
-    end
-  end
-
-  @spec await_source_ready() :: pid()
-  def await_source_ready() do
-    receive do
-      {:boombox_elixir_source, source} -> source
-    end
-  end
-
-  @spec await_sink_ready() :: pid()
-  def await_sink_ready() do
-    receive do
-      {:boombox_elixir_sink, sink} -> sink
-    end
-  end
-
   @impl true
   def handle_init(_ctx, opts) do
     spec =
@@ -73,9 +46,9 @@ defmodule Boombox.Pipeline do
     {[terminate: :normal], state}
   end
 
-  defp resolve_stream_endpoint({endpoint_type, opts}, parent)
+  defp resolve_elixir_endpoint({endpoint_type, opts}, parent)
        when endpoint_type in @elixir_endpoints,
        do: {endpoint_type, parent, opts}
 
-  defp resolve_stream_endpoint(endpoint, _parent), do: endpoint
+  defp resolve_elixir_endpoint(endpoint, _parent), do: endpoint
 end

lib/boombox/server.ex

@@ -401,7 +401,7 @@ defmodule Boombox.Server do
     %{supervisor: pipeline_supervisor, pipeline: pipeline} =
       boombox_opts
       |> Map.new()
-      |> Boombox.Pipeline.start_pipeline()
+      |> Boombox.Pipeline.start()
 
     {:noreply,
      %State{

python/src/boombox/boombox.py

@@ -168,22 +168,16 @@ def read(self) -> Generator[AudioPacket | VideoPacket, None, None]:
         RuntimeError
             If Boombox's output was not defined by an :py:class:`.RawData` endpoint.
         """
-        try:
-            while True:
-                match self._call(Atom("produce_packet")):
-                    case (Atom("ok"), packet):
-                        yield self._deserialize_packet(packet)
-                    case Atom("finished"):
-                        return
-                    case (Atom("error"), Atom("incompatible_mode")):
-                        raise RuntimeError(
-                            "Output not defined with an RawData endpoint."
-                        )
-                    case other:
-                        raise RuntimeError(f"Unknown response: {other}")
-        finally:
-            # pass
-            self.close()
+        while True:
+            match self._call(Atom("produce_packet")):
+                case (Atom("ok"), packet):
+                    yield self._deserialize_packet(packet)
+                case Atom("finished"):
+                    return
+                case (Atom("error"), Atom("incompatible_mode")):
+                    raise RuntimeError("Output not defined with an RawData endpoint.")
+                case other:
+                    raise RuntimeError(f"Unknown response: {other}")
 
     def write(self, packet: AudioPacket | VideoPacket) -> bool:
         """Write packets to Boombox.