Handle termination requests correctly

Noarkhh · Noarkhh · commit bd7808e862db · 2025-11-26T16:13:47.000+01:00
diff --git a/lib/boombox/server.ex b/lib/boombox/server.ex
@@ -111,8 +111,9 @@ defmodule Boombox.Server do
             membrane_source_demand: non_neg_integer(),
             pipeline_supervisor: pid() | nil,
             pipeline: pid() | nil,
-            ghosted_client: GenServer.from() | pid() | nil,
-            pipeline_exit_reason: term()
+            ghosted_client: GenServer.from() | Process.dest() | nil,
+            pipeline_termination_reason: term(),
+            termination_requested: boolean()
           }
 
     @enforce_keys [
@@ -131,7 +132,8 @@ defmodule Boombox.Server do
                   pipeline_supervisor: nil,
                   pipeline: nil,
                   ghosted_client: nil,
-                  pipeline_exit_reason: nil
+                  pipeline_termination_reason: nil,
+                  termination_requested: false
                 ]
   end
 
@@ -349,11 +351,16 @@ defmodule Boombox.Server do
 
     case state.communication_medium do
       :calls ->
-        if state.ghosted_client != nil do
-          reply(state.ghosted_client, :finished)
-          {:stop, reason, state}
-        else
-          {:noreply, %State{state | pipeline_exit_reason: reason}}
+        cond do
+          state.ghosted_client != nil ->
+            reply(state.ghosted_client, :finished)
+            {:stop, reason, state}
+
+          state.termination_requested ->
+            {:stop, reason, state}
+
+          true ->
+            {:noreply, %State{state | pipeline_termination_reason: reason}}
         end
 
       :messages ->
@@ -386,7 +393,7 @@ defmodule Boombox.Server do
     end
   end
 
-  @spec handle_request({:run, boombox_opts()}, GenServer.from() | pid(), State.t()) ::
+  @spec handle_request({:run, boombox_opts()}, GenServer.from() | Process.dest(), State.t()) ::
           {:reply, boombox_mode(), State.t()}
   defp handle_request({:run, boombox_opts}, from, state) do
     boombox_mode = get_boombox_mode(boombox_opts)
@@ -406,7 +413,7 @@ defmodule Boombox.Server do
      }}
   end
 
-  @spec handle_request(:get_pid, GenServer.from() | pid(), State.t()) ::
+  @spec handle_request(:get_pid, GenServer.from() | Process.dest(), State.t()) ::
           {:reply, pid(), State.t()}
   defp handle_request(:get_pid, _from, state) do
     {:reply, self(), state}
@@ -418,7 +425,7 @@ defmodule Boombox.Server do
 
   @spec handle_request(
           {:consume_packet, serialized_boombox_packet() | Boombox.Packet.t()},
-          GenServer.from() | pid(),
+          GenServer.from() | Process.dest(),
           State.t()
         ) ::
           {:reply, :ok | {:error, :incompatible_mode | :boombox_not_running}, State.t()}
@@ -438,8 +445,8 @@ defmodule Boombox.Server do
     state = %State{state | membrane_source_demand: state.membrane_source_demand - 1}
 
     cond do
-      state.pipeline_exit_reason != nil ->
-        {:stop, state.pipeline_exit_reason, :finished, state}
+      state.pipeline_termination_reason != nil ->
+        {:stop, state.pipeline_termination_reason, :finished, state}
 
       state.membrane_source_demand == 0 ->
         {:noreply, %State{state | ghosted_client: from}}
@@ -457,29 +464,29 @@ defmodule Boombox.Server do
     {:reply, {:error, :incompatible_mode}, state}
   end
 
-  @spec handle_request(:finish_consuming, GenServer.from() | pid(), State.t()) ::
-          {:reply, :ok | :finished | {:error, :incompatible_mode}, State.t()}
-          | {:stop, term(), :finished, State.t()}
+  @spec handle_request(:finish_consuming, GenServer.from() | Process.dest(), State.t()) ::
+          {:reply, :ok | {:error, :incompatible_mode}, State.t()}
+          | {:stop, term(), :ok, State.t()}
   defp handle_request(:finish_consuming, _from, %State{boombox_mode: :consuming} = state) do
-    if state.pipeline_exit_reason != nil do
-      {:stop, state.pipeline_exit_reason, :finished, state}
+    if state.pipeline_termination_reason != nil do
+      {:stop, state.pipeline_termination_reason, :ok, state}
     else
       send(state.membrane_source, {:boombox_eos, self()})
-      {:reply, :ok, state}
+      {:reply, :ok, %State{state | termination_requested: true}}
     end
   end
 
   defp handle_request(:finish_consuming, _from, %State{boombox_mode: _other_mode} = state) do
     {:reply, {:error, :incompatible_mode}, state}
   end
 
-  @spec handle_request(:produce_packet, GenServer.from() | pid(), State.t()) ::
+  @spec handle_request(:produce_packet, GenServer.from() | Process.dest(), State.t()) ::
           {:reply, {:error, :incompatible_mode | :boombox_not_running}, State.t()}
           | {:noreply, State.t()}
           | {:stop, term(), :finished, State.t()}
   defp handle_request(:produce_packet, from, %State{boombox_mode: :producing} = state) do
-    if state.pipeline_exit_reason != nil do
-      {:stop, state.pipeline_exit_reason, :finished, state}
+    if state.pipeline_termination_reason != nil do
+      {:stop, state.pipeline_termination_reason, :finished, state}
     else
       send(state.membrane_sink, {:boombox_demand, self()})
       {:noreply, %State{state | ghosted_client: from}}
@@ -490,30 +497,34 @@ defmodule Boombox.Server do
     {:reply, {:error, :incompatible_mode}, state}
   end
 
-  @spec handle_request(:finish_producing, GenServer.from() | pid(), State.t()) ::
+  @spec handle_request(:finish_producing, GenServer.from() | Process.dest(), State.t()) ::
           {:reply, :ok | {:error, :incompatible_mode}, State.t()}
   defp handle_request(:finish_producing, _from, %State{boombox_mode: :producing} = state) do
     Membrane.Pipeline.terminate(state.pipeline, asynchronous?: true)
-    {:reply, :ok, state}
+    {:reply, :ok, %State{state | termination_requested: true}}
   end
 
   defp handle_request(:finish_producing, _from, %State{boombox_mode: _other_mode} = state) do
     {:reply, {:error, :incompatible_mode}, state}
   end
 
-  @spec handle_request(term(), GenServer.from() | pid(), State.t()) ::
+  @spec handle_request(term(), GenServer.from() | Process.dest(), State.t()) ::
           {:reply, {:error, :invalid_request}, State.t()}
   defp handle_request(_invalid_request, _from, state) do
     {:reply, {:error, :invalid_request}, state}
   end
 
-  @spec reply(GenServer.from() | pid(), term()) :: :ok
-  defp reply(pid, reply_content) when is_pid(pid) do
-    send(pid, {:response, reply_content})
+  @spec reply(GenServer.from() | Process.dest(), term()) :: :ok
+  defp reply(dest, reply_content) when is_pid(dest) or is_port(dest) or is_atom(dest) do
+    send(dest, {:response, reply_content})
+  end
+
+  defp reply({name, node} = dest, reply_content) when is_atom(name) and is_atom(node) do
+    send(dest, {:response, reply_content})
   end
 
-  defp reply({pid, _tag} = client, reply_content) when is_pid(pid) do
-    GenServer.reply(client, reply_content)
+  defp reply({pid, _tag} = genserver_from, reply_content) when is_pid(pid) do
+    GenServer.reply(genserver_from, reply_content)
   end
 
   @spec get_boombox_mode(boombox_opts()) :: boombox_mode()
diff --git a/python/examples/anonymization_demo.py b/python/examples/anonymization_demo.py
@@ -26,6 +26,7 @@
 import queue
 import threading
 import time
+import logging
 from typing import NoReturn
 
 
@@ -66,6 +67,7 @@ def read_packets(boombox: Boombox, packet_queue: queue.Queue) -> None:
         packet_queue.put(packet)
         if packet_queue.qsize() > MAX_QUEUE_SIZE:
             packet_queue.get()
+    print("dupsko")
 
 
 def resize_frame(frame: np.ndarray, scale_factor: float) -> np.ndarray:
@@ -267,6 +269,9 @@ def main():
     SERVER_ADDRESS = "localhost"
     SERVER_PORT = 8000
 
+    logging.basicConfig()
+    Boombox.logger.setLevel(logging.INFO)
+
     threading.Thread(
         target=run_server, args=(SERVER_ADDRESS, SERVER_PORT), daemon=True
     ).start()
@@ -322,6 +327,7 @@ def main():
         args=(input_boombox, packet_queue),
         daemon=True,
     )
+    print("c")
     reading_thread.start()
     print("Input boombox initialized.")
 
@@ -363,7 +369,7 @@ def main():
             if should_anonymize:
                 packet.payload = distort_audio(packet.payload, packet.sample_rate)
 
-            output_boombox.write(packet)
+            print(output_boombox.write(packet))
 
         if isinstance(packet, VideoPacket):
             video_read_end_time = time.time() * 1000
@@ -398,7 +404,7 @@ def main():
                 render_transcription(transcription_lines, frame)
 
             packet.payload = frame
-            output_boombox.write(packet)
+            print(output_boombox.write(packet))
             video_read_start_time = time.time() * 1000
 
     output_boombox.close()
diff --git a/python/src/boombox/_vendor/pyrlang/node.py b/python/src/boombox/_vendor/pyrlang/node.py
@@ -288,7 +288,7 @@ def _send_local_registered(self, receiver, message) -> None:
                      receiver, receiver_obj, message)
             receiver_obj.deliver_message(msg=message)
         else:
-            LOG.warning("Send to unknown %s ignored", receiver)
+            LOG.info("Send to unknown %s ignored", receiver)
 
     def _send_local(self, receiver, message) -> None:
         """ Try find a process by pid and drop a message into its ``inbox_``.
diff --git a/python/src/boombox/_vendor/pyrlang/rex.py b/python/src/boombox/_vendor/pyrlang/rex.py
@@ -42,7 +42,7 @@ def handle_cast(self, msg):
 
     @info(1, lambda msg: True)
     def handle_info(self, msg):
-        LOG.error("rex unhandled info msg: %s", msg)
+        LOG.info("rex unhandled info msg: %s", msg)
 
 
 def act_on_msg(msg):
diff --git a/python/src/boombox/boombox.py b/python/src/boombox/boombox.py
@@ -29,7 +29,7 @@
 
 
 RELEASES_URL = "https://github.com/membraneframework/boombox/releases"
-PACKAGE_NAME = "boomboxlib"
+PACKAGE_NAME = "boomboxlibb"
 
 
 class Boombox(process.Process):
@@ -85,8 +85,15 @@ class Boombox(process.Process):
         Definition of an input or output of Boombox. Can be provided explicitly
         by an appropriate :py:class:`.BoomboxEndpoint` or a string of a path to
         a file or an URL, that Boombox will attempt to interpret as an endpoint.
+
+    Attributes
+    ----------
+    logger : ClassVar[logging.Logger]
+        Logger used in this class
     """
 
+    logger: ClassVar[logging.Logger]
+
     _python_node_name: ClassVar[str]
     _cookie: ClassVar[str]
 
@@ -105,6 +112,7 @@ class Boombox(process.Process):
     _python_node_name = f"{uuid.uuid4()}@127.0.0.1"
     _cookie = str(uuid.uuid4())
     _node = node.Node(node_name=_python_node_name, cookie=_cookie)
+    logger = logging.getLogger(__name__)
     threading.Thread(target=_node.run, daemon=True).start()
 
     def __init__(
@@ -131,7 +139,9 @@ def __init__(
         self._terminated = self.get_node().get_loop().create_future()
         self._finished = False
         self._receiver = (self._erlang_node_name, Atom("boombox_server"))
+        print("a")
         self._receiver = self._call(Atom("get_pid"))
+        print("b")
         self.get_node().monitor_process(self.pid_, self._receiver)
 
         boombox_arg = [
@@ -158,16 +168,22 @@ def read(self) -> Generator[AudioPacket | VideoPacket, None, None]:
         RuntimeError
             If Boombox's output was not defined by an :py:class:`.RawData` endpoint.
         """
-        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}")
+        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()
 
     def write(self, packet: AudioPacket | VideoPacket) -> bool:
         """Write packets to Boombox.
@@ -187,7 +203,7 @@ def write(self, packet: AudioPacket | VideoPacket) -> bool:
         Returns
         -------
         finished : bool
-            Informs if Boombox has finished accepting packets and closed its
+            If true then Boombox has finished accepting packets and closed its
             input for any further ones. Once it finishes processing the
             previously provided packet, it will terminate.
 
@@ -251,6 +267,7 @@ def close(self, wait: bool = True, kill: bool = False) -> None:
 
         match self._call(request):
             case Atom("ok"):
+                print("uuuu")
                 if kill:
                     self.kill()
                 elif wait:
@@ -284,8 +301,12 @@ def handle_one_inbox_message(self, msg: Any) -> None:
                 if not self._response.done():
                     self._response.set_result(response)
             case (Atom("DOWN"), _, Atom("process"), _, Atom("normal")):
+                print(self._boombox_mode)
+                print("1")
                 self._terminated.set_result(Atom("normal"))
             case (Atom("DOWN"), _, Atom("process"), _, reason):
+                print(self._boombox_mode)
+                print("2")
                 self._terminated.set_result(reason)
                 if not self._response.done():
                     self._response.set_exception(
@@ -342,9 +363,9 @@ def update_to(self, b=1, bsize=1, tsize=None):
         self._server_release_path = os.path.join(self._data_dir, "bin", "server")
 
         if os.path.exists(self._server_release_path):
-            logging.info("Elixir boombox release already present.")
+            self.logger.info("Elixir boombox release already present.")
             return
-        logging.info("Elixir boombox release not found, downloading...")
+        self.logger.info("Elixir boombox release not found, downloading...")
 
         if self._version == "dev":
             release_url = os.path.join(RELEASES_URL, "latest/download")
@@ -369,13 +390,13 @@ def update_to(self, b=1, bsize=1, tsize=None):
             unit_scale=True,
             unit_divisor=1024,
             miniters=1,
-            desc=f"Downloading {release_tarball}",
+            desc=f"Downloading {release_tarball} from {release_url}",
         ) as t:
             urllib.request.urlretrieve(
                 download_url, filename=tarball_path, reporthook=t.update_to
             )
 
-        logging.info("Download complete. Extracting...")
+        self.logger.info("Download complete. Extracting...")
         with tarfile.open(tarball_path) as tar:
             tar.extractall(self._data_dir)
             os.remove(tarball_path)
