Repository: livebook-dev/kino
Pull request #268 (2023-03-17 12:24:18)

Add concurrent listen
Adds a version of `listen/2` that processes all events concurrently in separate processes.


Patch:
diff --git a/lib/kino.ex b/lib/kino.ex
index 04bf713..0f39dad 100644
--- a/lib/kino.ex
+++ b/lib/kino.ex
@@ -329,7 +329,10 @@ defmodule Kino do
   end
 
   @doc ~S"""
-  Asynchronously consumes a stream with `fun`.
+  Consumes a stream with `fun` without blocking execution.
+
+  Note that events are processed by `fun` sequentially. If you want
+  to process them concurrently, use `async_listen/2`.
 
   ## Examples
 
@@ -443,6 +446,44 @@ defmodule Kino do
     :ok
   end
 
+  @doc """
+  Same as `listen/2`, except each event is processed concurrently.
+  """
+  @spec async_listen(Enumerable.t() | pos_integer(), (term() -> any())) :: :ok
+  def async_listen(stream_or_interval_ms, fun)
+
+  def async_listen(interval_ms, fun) when is_integer(interval_ms) and is_function(fun, 1) do
+    async_listen(Stream.interval(interval_ms), fun)
+  end
+
+  def async_listen(stream, fun) when is_function(fun, 1) do
+    async(fn ->
+      # For organization purposes we start all tasks under a separate
+      # supervisor and only that supervisor is started with Kino.start_child/1
+
+      start_fun = fn ->
+        {:ok, task_supervisor} = start_child(Task.Supervisor)
+        task_supervisor
+      end
+
+      reducer = fn event, task_supervisor ->
+        {[{event, task_supervisor}], task_supervisor}
+      end
+
+      after_fun = fn task_supervisor ->
+        DynamicSupervisor.terminate_child(Kino.DynamicSupervisor, task_supervisor)
+      end
+
+      stream
+      |> Stream.transform(start_fun, reducer, after_fun)
+      |> Enum.each(fn {event, task_supervisor} ->
+        Task.Supervisor.start_child(task_supervisor, fn ->
+          safe_apply(fun, [event], "Kino.async_listen")
+        end)
+      end)
+    end)
+  end
+
   @doc """
   Returns a special value that results in no visible output.
   """


Test patch:
diff --git a/test/kino_test.exs b/test/kino_test.exs
index 79e5962..6c36721 100644
--- a/test/kino_test.exs
+++ b/test/kino_test.exs
@@ -85,7 +85,7 @@ defmodule KinoTest do
   end
 
   describe "listen/2" do
-    test "asynchronously consumes stream items" do
+    test "consumes stream items" do
       myself = self()
 
       Stream.iterate(0, &(&1 + 1))
@@ -140,7 +140,7 @@ defmodule KinoTest do
   end
 
   describe "listen/3" do
-    test "asynchronously consumes stream items and accumulates state" do
+    test "consumes stream items and accumulates state" do
       myself = self()
 
       Stream.iterate(0, &(&1 + 1))
@@ -198,6 +198,65 @@ defmodule KinoTest do
     end
   end
 
+  describe "async_listen/2" do
+    test "concurrently processes stream items" do
+      myself = self()
+
+      Stream.iterate(0, &(&1 + 1))
+      |> Stream.take(2)
+      |> Kino.async_listen(fn i ->
+        send(myself, {:item, i})
+        Process.sleep(:infinity)
+      end)
+
+      assert_receive {:item, 0}
+      assert_receive {:item, 1}
+    end
+
+    test "with control events" do
+      button = Kino.Control.button("Click")
+      myself = self()
+      trace_subscription()
+
+      Kino.async_listen(button, fn event ->
+        send(myself, event)
+        Process.sleep(:infinity)
+      end)
+
+      assert_receive {:trace, _, :receive, {:"$gen_cast", {:subscribe, ref, _, _}}}
+                     when button.attrs.ref == ref
+
+      info = %{origin: "client1"}
+      send(button.attrs.destination, {:event, button.attrs.ref, info})
+      send(button.attrs.destination, {:event, button.attrs.ref, info})
+
+      assert_receive ^info
+      assert_receive ^info
+    end
+
+    test "ignores failures" do
+      log =
+        capture_log(fn ->
+          myself = self()
+
+          Stream.iterate(0, &(&1 + 1))
+          |> Stream.take(2)
+          |> Kino.async_listen(fn i ->
+            send(myself, {:item, self()})
+            1 = i
+          end)
+
+          assert_receive {:item, pid1}
+          assert_receive {:item, pid2}
+          await_process(pid1)
+          await_process(pid2)
+        end)
+
+      assert log =~ "Kino.async_listen"
+      assert log =~ "** (MatchError) no match of right hand side value: 0"
+    end
+  end
+
   defp await_process(pid) do
     ref = Process.monitor(pid)
     assert_receive {:DOWN, ^ref, :process, _object, _reason}