Refactor elixir endpoints #118

Noarkhh · 2025-11-14T17:22:22Z

When using server-based endpoints (reader, writer, message), the server will now spawn the pipeline directly and communicate with elixir elements directly.

I'm not sure which approach is better for having push and pull elixir source and sink - using defdelegetes and have a common behavior module (the current one) or have a macro that generates push and pull element.

mat-hek · 2025-11-28T15:13:14Z

@Noarkhh I'd go for a macro that generates push & pull elements ;)

mat-hek · 2025-12-01T11:11:31Z

lib/boombox/server.ex

+            membrane_source_demand: non_neg_integer(),
+            pipeline_supervisor: pid() | nil,
+            pipeline: pid() | nil,
+            ghosted_client: GenServer.from() | Process.dest() | nil,


This name is a bit ambiguous, maybe client_to_reply or current_client?

mat-hek · 2025-12-01T11:14:55Z

python/src/boombox/boombox.py


 RELEASES_URL = "https://github.com/membraneframework/boombox/releases"
-PACKAGE_NAME = "boomboxlib"
+PACKAGE_NAME = "boomboxlibb"


mat-hek · 2025-12-01T11:17:41Z

python/src/boombox/boombox.py

+
+        match self._call(request):
+            case Atom("ok"):
+                print("uuuu")


mat-hek · 2025-12-01T11:20:40Z

lib/boombox/pipeline.ex

+      |> Map.update!(:input, &resolve_elixir_endpoint(&1, self()))
+      |> Map.update!(:output, &resolve_elixir_endpoint(&1, self()))


endpoint resolution happens in the internal bin, can we move this there?

not really, because we need to call self() in the parent process, not the bin

We're passing this pid as a parent option. We can pass it to the internal bin as well

mat-hek · 2025-12-01T11:32:26Z

examples.livemd

 Stream.unfold(%{}, fn _state ->
-  {result1, packet1} = Boombox.read(reader1)
-  {result2, packet2} = Boombox.read(reader2)
+  case {Boombox.read(reader1), Boombox.read(reader2)} do
+    {:finished, :finished} ->
+      nil

-  joined_image =
-    Vix.Vips.Operation.join!(packet1.payload, packet2.payload, :VIPS_DIRECTION_HORIZONTAL)
+    {{:ok, _packet}, :finished} ->
+      Boombox.close(reader1)
+      nil

-  packet = %Boombox.Packet{
-    pts: max(packet1.pts, packet2.pts),
-    payload: joined_image,
-    kind: :video
-  }
+    {:finished, {:ok, _packet}} ->
+      Boombox.close(reader2)
+      nil

-  Boombox.write(writer, packet)
+    {{:ok, packet1}, {:ok, packet2}} ->
+      joined_image =
+        Vix.Vips.Operation.join!(packet1.payload, packet2.payload, :VIPS_DIRECTION_HORIZONTAL)

-  if :finished in [result1, result2] do
-    if result1 == :ok, do:
-      Boombox.close(reader1)
-    if result2 == :ok, do:
-      Boombox.close(reader2)
-    nil
-  else
-    {nil, %{}}
+      packet = %Boombox.Packet{
+        pts: max(packet1.pts, packet2.pts),
+        payload: joined_image,
+        kind: :video
+      }
+
+      Boombox.write(writer, packet)
+
+      {nil, %{}}
  end


Stream.unfold with an ignored state looks weird. Also, matching on all the possibilities seems redundant. I'd go for

Suggested change

Stream.unfold(%{}, fn _state ->

{result1, packet1} = Boombox.read(reader1)

{result2, packet2} = Boombox.read(reader2)

case {Boombox.read(reader1), Boombox.read(reader2)} do

{:finished, :finished} ->

nil

joined_image =

Vix.Vips.Operation.join!(packet1.payload, packet2.payload, :VIPS_DIRECTION_HORIZONTAL)

{{:ok, _packet}, :finished} ->

Boombox.close(reader1)

nil

packet = %Boombox.Packet{

pts: max(packet1.pts, packet2.pts),

payload: joined_image,

kind: :video

}

{:finished, {:ok, _packet}} ->

Boombox.close(reader2)

nil

Boombox.write(writer, packet)

{{:ok, packet1}, {:ok, packet2}} ->

joined_image =

Vix.Vips.Operation.join!(packet1.payload, packet2.payload, :VIPS_DIRECTION_HORIZONTAL)

if :finished in [result1, result2] do

if result1 == :ok, do:

Boombox.close(reader1)

if result2 == :ok, do:

Boombox.close(reader2)

nil

else

{nil, %{}}

packet = %Boombox.Packet{

pts: max(packet1.pts, packet2.pts),

payload: joined_image,

kind: :video

}

Boombox.write(writer, packet)

{nil, %{}}

end

Stream.repeatedly(fn ->

case {Boombox.read(reader1), Boombox.read(reader2)} do

{{: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

}

:ok = Boombox.write(writer, packet)

_finished ->

:eos

end

end)

|> Enum.find(& &1 == :eos)

@mat-hek it looks good we just need to remember to call Boombox.close()

Right, missed that. Maybe Boombox.close on already closed boombox could be a noop, so we could just close both boomboxes no matter what?

varsill · 2025-12-02T14:42:31Z

lib/boombox/internal_bin/elixir_endpoints/pull_sink.ex

+
+  alias Boombox.InternalBin.ElixirEndpoints.Sink
+
+  def_input_pad :input,


How many pads of this type can exist? Wouldn't max_instances be applicable here?

varsill · 2025-12-02T14:45:01Z

examples.livemd

+       bb_states: %{
         bb1: %{last_packet: nil, eos: false},
         bb2: %{last_packet: nil, eos: false}
       },
-       input_boomboxes: %{bb1 => :bb1, bb2 => :bb2},
+       bbs: %{bb1 => :bb1, bb2 => :bb2},


[NIT] I am not 100% encouraged by this change, bbs look quite funny to me :D

varsill · 2025-12-02T14:47:56Z

examples.livemd

 Stream.unfold(%{}, fn _state ->
-  {result1, packet1} = Boombox.read(reader1)
-  {result2, packet2} = Boombox.read(reader2)
+  case {Boombox.read(reader1), Boombox.read(reader2)} do
+    {:finished, :finished} ->
+      nil

-  joined_image =
-    Vix.Vips.Operation.join!(packet1.payload, packet2.payload, :VIPS_DIRECTION_HORIZONTAL)
+    {{:ok, _packet}, :finished} ->
+      Boombox.close(reader1)
+      nil

-  packet = %Boombox.Packet{
-    pts: max(packet1.pts, packet2.pts),
-    payload: joined_image,
-    kind: :video
-  }
+    {:finished, {:ok, _packet}} ->
+      Boombox.close(reader2)
+      nil

-  Boombox.write(writer, packet)
+    {{:ok, packet1}, {:ok, packet2}} ->
+      joined_image =
+        Vix.Vips.Operation.join!(packet1.payload, packet2.payload, :VIPS_DIRECTION_HORIZONTAL)

-  if :finished in [result1, result2] do
-    if result1 == :ok, do:
-      Boombox.close(reader1)
-    if result2 == :ok, do:
-      Boombox.close(reader2)
-    nil
-  else
-    {nil, %{}}
+      packet = %Boombox.Packet{
+        pts: max(packet1.pts, packet2.pts),
+        payload: joined_image,
+        kind: :video
+      }
+
+      Boombox.write(writer, packet)
+
+      {nil, %{}}
  end


@mat-hek it looks good we just need to remember to call Boombox.close()

varsill · 2025-12-02T14:56:57Z

test/boombox_test.exs

            Stream.unfold(:ok, fn
              :ok ->
-                {result, packet} = Boombox.read(boombox)
-                {packet, result}
-
-              :finished ->
-                nil
+                case Boombox.read(boombox) do
+                  {:ok, packet} -> {packet, :ok}
+                  :finished -> nil
+                end


[NIT] This "state" here is useless so you can remove it
And similarly to the case in examples.livemd, you could use Enum.repeatedly here

varsill · 2025-12-02T14:59:58Z

python/src/boombox/boombox.py

                if not self._response.done():
                    self._response.set_result(response)
            case (Atom("DOWN"), _, Atom("process"), _, Atom("normal")):
+                print(self._boombox_mode)


I think you can remove these :D

varsill · 2025-12-02T15:05:13Z

lib/boombox/internal_bin/elixir_endpoints/sink.ex


-  @impl true
-  def handle_info(:boombox_demand, _ctx, state) do
+  def handle_info({:boombox_demand, consumer}, _ctx, %{consumer: consumer} = state) do


shouldn't we buffer that demand up if consumer is not yet available?

Actually after looking at the whole module I am not sure if the consumer field in the state is defined anywhere

the consumer is passed with options, but I agree, this needs better typing

varsill · 2025-12-02T15:08:59Z

lib/boombox/internal_bin/elixir_endpoints/push_sink.ex

+  def_input_pad :input,
+    accepted_format: any_of(Membrane.RawAudio, Membrane.RawVideo),
+    availability: :on_request,
+    flow_control: :auto


Why :auto instead of :push?

varsill · 2025-12-10T08:45:31Z

lib/boombox/internal_bin/elixir_endpoints/sink.ex

TBH I like the version with defdelegate more, but let it be :D

I'll add a comment 🥰

varsill · 2025-12-10T08:49:18Z

lib/boombox/internal_bin/elixir_endpoints/sink.ex

+    def_options(
+      consumer: [
+        spec: pid()
+      ]
+    )


You can remove the parentheses.
I would also opt for some :description (just to mention that it's a pid of a process that will receive the {:boombox_packet, ...} messages) even though it's a private module :D

varsill · 2025-12-11T13:11:38Z

lib/boombox/internal_bin/elixir_endpoints/source.ex

+#     handles demands from subsequent element by demanding packets from the producer
+#     process with `{:boombox_demand, self(), demand_amount}` messages.
+#   * PushSource - The element has `:push` flow control on output pads and expects
+#     the producer process to provide it packets without demanding them.


Suggested change

# the producer process to provide it packets without demanding them.

# the producer process to provide it with packets without demanding them.

varsill · 2025-12-11T13:12:40Z

examples.livemd


 # MIX_INSTALL_CONFIG_BEGIN
-boombox = {:boombox, github: "membraneframework/boombox"}
+boombox = {:boombox, github: "membraneframework/boombox", branch: "refactor-elixir-endpoints"}


remember to update it before merging

varsill · 2025-12-11T13:17:20Z

lib/boombox.ex

  Returns `:ok` if more packets can be provided, and
  `:finished` when Boombox finished consuming and will not accept any more packets. Returns
-  synchronously once the packet has been processed by Boombox.
+  synchronously once the packet has been ingested Boombox is ready for more packets.


Suggested change

synchronously once the packet has been ingested Boombox is ready for more packets.

synchronously once the packet has been ingested and Boombox is ready for more packets.

varsill · 2025-12-11T14:20:37Z

lib/boombox/internal_bin/elixir_endpoints/source.ex

+      def handle_demand(Pad.ref(:output, _id), _size, _unit, ctx, state) do
+        demands = Enum.map(ctx.pads, fn {_pad, %{demand: demand}} -> demand end)
+
+        if Enum.all?(demands, &(&1 > 0)) do
+          send(state.producer, {:boombox_demand, self(), Enum.sum(demands)})
+        end
+
+        {[], state}
+      end


As discussed, handling demand requires changes

mat-hek · 2025-12-19T11:34:46Z

lib/boombox.ex

+
+  # value, _state ->
+  # raise ArgumentError, "Expected Boombox.Packet.t(), got: #{inspect(value)}"
+  # end


mat-hek · 2025-12-23T10:23:48Z

lib/boombox.ex

+      packets and will begin terminating. No more messages will be sent.

  ```
  Boombox.run(


This example doesn't fit here

mat-hek · 2025-12-23T10:24:18Z

lib/boombox.ex

-
-  If the output is a `:reader` endpoint this function will return a `Boombox.Reader` struct,
-  which is used to read media packets from boombox with `read/1`.
+  Input endpoints with special behaviours:


Let's add one sentence describing what's a 'normal' behaviour

mat-hek · 2025-12-23T10:55:32Z

lib/boombox/internal_bin/elixir_endpoints/source.ex

+      def handle_demand(Pad.ref(:output, id), size, _unit, _ctx, state) do
+        send(state.producer, {:boombox_demand, self(), id, size})
+        {[], state}


With this approach, don't we have the same problem as before:

handle_demand is called

we send the demand to boombox

boombox receives the demand

handle_demand is called again, with bigger demand

boombox sends buffers to element

boombox receives the new demand and saves it, but the actual demand is smaller, as buffers were sent

We could use ctx.incoming_demand, send that to boombox and sum it in boombox to avoid this problem

Noarkhh self-assigned this Nov 14, 2025

Noarkhh added this to Smackore Nov 14, 2025

Noarkhh moved this to In Progress in Smackore Nov 14, 2025

Noarkhh changed the base branch from master to message-endpoint November 14, 2025 17:23

Noarkhh force-pushed the refactor-elixir-endpoints branch from 7cb399c to e85f6a1 Compare November 14, 2025 17:23

Noarkhh force-pushed the message-endpoint branch 3 times, most recently from 88a148a to e44efe8 Compare November 26, 2025 15:17

Noarkhh force-pushed the refactor-elixir-endpoints branch 2 times, most recently from 9ac398d to b729d70 Compare November 26, 2025 15:20

Noarkhh requested review from mat-hek and varsill November 26, 2025 15:54

Noarkhh force-pushed the refactor-elixir-endpoints branch 2 times, most recently from 4c07d31 to a053f6c Compare November 26, 2025 15:59

Noarkhh moved this from In Progress to In Review in Smackore Nov 26, 2025

mat-hek requested changes Dec 1, 2025

View reviewed changes

varsill requested changes Dec 2, 2025

View reviewed changes

Noarkhh force-pushed the refactor-elixir-endpoints branch 5 times, most recently from b46ae53 to e99f2d6 Compare December 3, 2025 17:05

Noarkhh requested review from mat-hek and varsill December 4, 2025 15:21

Noarkhh added 5 commits December 10, 2025 09:36

Implement a message endpoint

94272ab

Fix typing

b3906aa

Add tests

67934c5

Add demos

025af1d

Allow for manual finish of reading

110ffa6

Noarkhh added 2 commits December 10, 2025 09:36

Add docs

6c38c3b

Remove opaque types to satisfy dialyzer

9ffaab6

Noarkhh force-pushed the message-endpoint branch from e44efe8 to 9ffaab6 Compare December 10, 2025 08:37

Noarkhh added 2 commits December 10, 2025 10:00

Update deps

3cb1362

Fix warnings

1845875

varsill reviewed Dec 11, 2025

View reviewed changes

Noarkhh force-pushed the refactor-elixir-endpoints branch from e99f2d6 to e680f73 Compare December 11, 2025 11:01

Noarkhh added 16 commits December 11, 2025 12:05

Implement a message endpoint

9011e3f

refactor wip

233c155

WIP

e4bf357

wip

a946dd9

Make the refactor work for pull flow control

2e11129

Refactor working

9319e26

Fix receiving finished

8f9fb5d

Fix typespecs

93524de

Fix sink

7f15c5b

Handle termination requests correctly

a4c8e9b

Fix examples

e99d9d0

Fix naming

ce0161b

Apply reviewers suggestions

bb51710

Add max_instances to elements

bac892a

Satisfy 1.19 compiler

79dc568

Improve documentation of Elixir Sink and Source

d71885f

Noarkhh force-pushed the refactor-elixir-endpoints branch from e680f73 to d71885f Compare December 11, 2025 11:14

Noarkhh requested a review from varsill December 11, 2025 11:18

varsill requested changes Dec 11, 2025

View reviewed changes

Add more tests

be1bb81

Noarkhh requested a review from varsill December 18, 2025 11:12

Base automatically changed from message-endpoint to master December 18, 2025 11:44

mat-hek reviewed Dec 23, 2025

View reviewed changes

		\|> Map.update!(:input, &resolve_elixir_endpoint(&1, self()))
		\|> Map.update!(:output, &resolve_elixir_endpoint(&1, self()))


		alias Boombox.InternalBin.ElixirEndpoints.Sink

		def_input_pad :input,

	# the producer process to provide it packets without demanding them.
	# the producer process to provide it with packets without demanding them.

	synchronously once the packet has been ingested Boombox is ready for more packets.
	synchronously once the packet has been ingested and Boombox is ready for more packets.

Refactor elixir endpoints #118

Are you sure you want to change the base?

Refactor elixir endpoints #118

Uh oh!

Conversation

Noarkhh commented Nov 14, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mat-hek commented Nov 28, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Noarkhh commented Nov 14, 2025 •

edited

Loading