sse: use stream_split dep rather than half-assed vendoring.

Author: thruflo

packages/sync-service/lib/electric/shapes/api.ex

@@ -700,7 +700,7 @@ defmodule Electric.Shapes.Api do
     #
     # The JSON encoder groups stream elements by 500. So perhaps this
     # could be a larger number for more efficiency?
-    case Electric.Utils.take_and_drop(stream, 3) do
+    case StreamSplit.take_and_drop(stream, 3) do
       {[], _tail} ->
         {[], %{state | mode: :receive, stream: nil}}
 

packages/sync-service/lib/electric/utils.ex

@@ -717,85 +717,4 @@ defmodule Electric.Utils do
       after_fun
     )
   end
-
-  @doc """
-  This function is a combination of `Enum.take/2` and `Enum.drop/2` returning
-  first `n` dropped elements and the rest of the enum as a stream.
-
-  The important difference is that the enumerable is only iterated once, and
-  only for the required `n` items. The rest of the enumerable may be iterated
-  lazily later from the returned stream.
-
-  ## Examples
-      iex> {head, tail} = take_and_drop(Stream.cycle(1..3), 4)
-      iex> head
-      [1, 2, 3, 1]
-      iex> Enum.take(tail, 7)
-      [2, 3, 1, 2, 3, 1, 2]
-
-  Cribbed from https://github.com/tallakt/stream_split
-  """
-  @spec take_and_drop(Enumerable.t(), pos_integer) :: {List.t(), Enumerable.t()}
-  def take_and_drop(enum, n) when n > 0 do
-    case apply_reduce(enum, n) do
-      {:done, {_, list}} ->
-        {:lists.reverse(list), []}
-
-      {:suspended, {_, list}, cont} ->
-        stream_split = %{continuation: cont, stream: continuation_to_stream(cont)}
-        {:lists.reverse(list), stream_split}
-
-      {:halted, {_, list}} ->
-        {:lists.reverse(list), []}
-    end
-  end
-
-  def take_and_drop(enum, 0) do
-    {[], enum}
-  end
-
-  defp apply_reduce(%{continuation: cont}, n) do
-    cont.({:cont, {n, []}})
-  end
-
-  defp apply_reduce(enum, n) do
-    Enumerable.reduce(enum, {:cont, {n, []}}, &reducer_helper/2)
-  end
-
-  defp reducer_helper(item, :tail) do
-    {:suspend, item}
-  end
-
-  defp reducer_helper(item, {c, list}) when c > 1 do
-    {:cont, {c - 1, [item | list]}}
-  end
-
-  defp reducer_helper(item, {_, list}) do
-    {:suspend, {0, [item | list]}}
-  end
-
-  defp continuation_to_stream(cont) do
-    wrapped = fn {_, _, acc_cont} ->
-      case acc_cont.({:cont, :tail}) do
-        acc = {:suspended, item, _cont} ->
-          {[item], acc}
-
-        {:halted, acc} ->
-          {:halt, acc}
-
-        {:done, acc} ->
-          {:halt, acc}
-      end
-    end
-
-    cleanup = fn
-      {:suspended, _, acc_cont} ->
-        acc_cont.({:halt, nil})
-
-      _ ->
-        nil
-    end
-
-    Stream.resource(fn -> {:suspended, nil, cont} end, wrapped, cleanup)
-  end
 end

packages/sync-service/mix.exs

@@ -103,6 +103,7 @@ defmodule Electric.MixProject do
         {:retry, "~> 0.18"},
         {:remote_ip, "~> 1.2"},
         {:req, "~> 0.5"},
+        {:stream_split, "~> 0.1"},
         {:telemetry_poller, "~> 1.1"},
         {:tls_certificate_check, "~> 1.23"},
         {:tz, "~> 0.27"}

packages/sync-service/mix.lock

@@ -52,6 +52,7 @@
   "sentry": {:hex, :sentry, "10.8.0", "1e8cc0ef21401e5914e6fc2f37489d6c685d31a0556dbd8ab4709cc1587a7232", [:mix], [{:hackney, "~> 1.8", [hex: :hackney, repo: "hexpm", optional: true]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: true]}, {:nimble_options, "~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:nimble_ownership, "~> 0.3.0 or ~> 1.0", [hex: :nimble_ownership, repo: "hexpm", optional: false]}, {:phoenix, "~> 1.6", [hex: :phoenix, repo: "hexpm", optional: true]}, {:phoenix_live_view, "~> 0.20", [hex: :phoenix_live_view, repo: "hexpm", optional: true]}, {:plug, "~> 1.6", [hex: :plug, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: true]}], "hexpm", "92549e7ba776b7ccfed4e74d58987272d37d99606b130e4141bc015a1a8e4235"},
   "ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.7", "354c321cf377240c7b8716899e182ce4890c5938111a1296add3ec74cf1715df", [:make, :mix, :rebar3], [], "hexpm", "fe4c190e8f37401d30167c8c405eda19469f34577987c76dde613e838bbc67f8"},
   "stream_data": {:hex, :stream_data, "1.1.3", "15fdb14c64e84437901258bb56fc7d80aaf6ceaf85b9324f359e219241353bfb", [:mix], [], "hexpm", "859eb2be72d74be26c1c4f272905667672a52e44f743839c57c7ee73a1a66420"},
+  "stream_split": {:hex, :stream_split, "0.1.7", "2d3fd1fd21697da7f91926768d65f79409086052c9ec7ae593987388f52425f8", [:mix], [], "hexpm", "1dc072ff507a64404a0ad7af90df97096183fee8eeac7b300320cea7c4679147"},
   "telemetry": {:hex, :telemetry, "1.3.0", "fedebbae410d715cf8e7062c96a1ef32ec22e764197f70cda73d82778d61e7a2", [:rebar3], [], "hexpm", "7015fc8919dbe63764f4b4b87a95b7c0996bd539e0d499be6ec9d7f3875b79e6"},
   "telemetry_metrics": {:hex, :telemetry_metrics, "1.1.0", "5bd5f3b5637e0abea0426b947e3ce5dd304f8b3bc6617039e2b5a008adc02f8f", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "e7b79e8ddfde70adb6db8a6623d1778ec66401f366e9a8f5dd0955c56bc8ce67"},
   "telemetry_metrics_prometheus_core": {:hex, :telemetry_metrics_prometheus_core, "1.2.1", "c9755987d7b959b557084e6990990cb96a50d6482c683fb9622a63837f3cd3d8", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:telemetry_metrics, "~> 0.6 or ~> 1.0", [hex: :telemetry_metrics, repo: "hexpm", optional: false]}], "hexpm", "5e2c599da4983c4f88a33e9571f1458bf98b0cf6ba930f1dc3a6e8cf45d5afb6"},