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

@@ -212,12 +212,12 @@ defmodule Boombox do
     case opts do
       %{input: {:stream, _stream_opts}} ->
         procs = Pipeline.start_pipeline(opts)
-        source = Pipeline.await_source_ready()
+        source = await_source_ready()
         consume_stream(stream, source, procs)
 
       %{output: {:stream, _stream_opts}} ->
         procs = Pipeline.start_pipeline(opts)
-        sink = Pipeline.await_sink_ready()
+        sink = await_sink_ready()
         produce_stream(sink, procs)
 
       %{input: {:writer, _writer_opts}} ->
@@ -237,7 +237,7 @@ defmodule Boombox do
       opts ->
         opts
         |> Pipeline.start_pipeline()
-        |> Pipeline.await_pipeline()
+        |> await_pipeline()
     end
   end
 
@@ -277,7 +277,7 @@ defmodule Boombox do
     case opts do
       %{input: {:stream, _stream_opts}} ->
         procs = Pipeline.start_pipeline(opts)
-        source = Pipeline.await_source_ready()
+        source = await_source_ready()
 
         Task.async(fn ->
           Process.monitor(procs.supervisor)
@@ -286,7 +286,7 @@ defmodule Boombox do
 
       %{output: {:stream, _stream_opts}} ->
         procs = Pipeline.start_pipeline(opts)
-        sink = Pipeline.await_sink_ready()
+        sink = await_sink_ready()
         produce_stream(sink, procs)
 
       %{input: {:writer, _writer_opts}} ->
@@ -311,7 +311,7 @@ defmodule Boombox do
         task =
           Task.async(fn ->
             Process.monitor(procs.supervisor)
-            Pipeline.await_pipeline(procs)
+            await_pipeline(procs)
           end)
 
         await_external_resource_ready()
@@ -322,7 +322,7 @@ defmodule Boombox do
 
         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

@@ -13,8 +13,8 @@ defmodule Boombox.Pipeline do
   def start_pipeline(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