elixir-client: Fix consumption of client streams across processes (#2798)

Fixes #2797

Author: magnetised

.changeset/tidy-teachers-fetch.md

@@ -0,0 +1,5 @@
+---
+"@core/elixir-client": patch
+---
+
+Fix consumption of embedded API across processes

packages/elixir-client/lib/electric/client/embedded.ex

@@ -47,7 +47,9 @@ if Code.ensure_loaded?(Electric.Shapes.Api) do
         schema: Api.schema(response),
         next_cursor: nil,
         request_timestamp: timestamp,
-        body: response.body
+        # Ensure we enumerate the response body in the same process that made the
+        # original api request so the cleanup shape process cleanup works
+        body: Enum.into(response.body, [])
       }
     end
 

packages/elixir-client/test/electric/client/embedded_test.exs

@@ -202,4 +202,24 @@ defmodule Electric.Client.EmbeddedTest do
       _msgs = stream(%{ctx | shape: shape}, 4)
     end
   end
+
+  test "is able to read response body in separate processes", ctx do
+    parent = self()
+
+    {:ok, _id1} = insert_item(ctx)
+    {:ok, _id2} = insert_item(ctx)
+    {:ok, _id3} = insert_item(ctx)
+
+    stream = stream(ctx, live: false)
+
+    {:ok, _} =
+      Task.start_link(fn ->
+        Enum.each(stream, &send(parent, {:msg, &1}))
+      end)
+
+    assert_receive {:msg, %ChangeMessage{}}, 1000
+    assert_receive {:msg, %ChangeMessage{}}, 1000
+    assert_receive {:msg, %ChangeMessage{}}, 1000
+    assert_receive {:msg, up_to_date()}, 1000
+  end
 end