Refactor, add specs

Author: antonmi

TODO.md

@@ -13,4 +13,6 @@ Clustering:
 
 
 Refactoring
-- chunk transformer
+- chunk transformer
+- hanging streams after delete component 
+- empty_gen_mix at the end of composite. delete, insert, and replace the last component

lib/composite.ex

@@ -99,35 +99,47 @@ defmodule Strom.Composite do
   @spec stop(__MODULE__.t()) :: :ok
   def stop(%__MODULE__{} = composite), do: StartStop.stop(composite)
 
+  @spec delete(__MODULE__.t(), {integer(), integer()}) :: __MODULE__.t()
   def delete(composite, {index_from, index_to}) do
     GenServer.call(composite.name, {:delete, index_from, index_to})
   end
 
+  @spec delete(__MODULE__.t(), integer()) :: __MODULE__.t()
   def delete(composite, index) do
     delete(composite, {index, index})
   end
 
+  @spec insert(__MODULE__.t(), integer(), Strom.component()) :: {__MODULE__.t(), Strom.flow()}
   def insert(composite, index, new_component) when is_struct(new_component) do
     insert(composite, index, [new_component])
   end
 
+  @spec insert(__MODULE__.t(), integer(), list(Strom.component())) ::
+          {__MODULE__.t(), Strom.flow()}
   def insert(composite, index, new_components) when is_list(new_components) do
     GenServer.call(composite.name, {:insert, index, new_components})
   end
 
+  @spec replace(__MODULE__.t(), integer(), Strom.component()) :: {__MODULE__.t(), Strom.flow()}
   def replace(composite, index, new_component)
       when is_integer(index) and is_struct(new_component) do
     replace(composite, {index, index}, [new_component])
   end
 
+  @spec replace(__MODULE__.t(), {integer(), integer()}, Strom.component()) ::
+          {__MODULE__.t(), Strom.flow()}
   def replace(composite, {index_from, index_to}, new_component) when is_struct(new_component) do
     replace(composite, {index_from, index_to}, [new_component])
   end
 
+  @spec replace(__MODULE__.t(), {integer(), integer()}, list(Strom.component())) ::
+          {__MODULE__.t(), Strom.flow()}
   def replace(composite, {index_from, index_to}, new_components) when is_list(new_components) do
     GenServer.call(composite.name, {:replace, {index_from, index_to}, new_components})
   end
 
+  @spec replace(__MODULE__.t(), integer(), list(Strom.component())) ::
+          {__MODULE__.t(), Strom.flow()}
   def replace(composite, index, new_components)
       when is_integer(index) and is_list(new_components) do
     replace(composite, {index, index}, new_components)
@@ -155,9 +167,11 @@ defmodule Strom.Composite do
   def handle_call(
         {:delete, index_from, index_to},
         _from,
-        %__MODULE__{components: components} = composite
+        %__MODULE__{components: components, name: name} = composite
       ) do
-    {components, _deleted_components} = Manipulations.delete(components, index_from, index_to)
+    {components, _deleted_components, %{}} =
+      Manipulations.replace(components, index_from, index_to, [], name)
+
     composite = %{composite | components: components}
     {:reply, composite, composite}
   end
@@ -168,7 +182,7 @@ defmodule Strom.Composite do
         %__MODULE__{components: components, name: name} = composite
       )
       when is_list(new_components) do
-    {components, subflow} = Manipulations.insert(components, index, new_components, name)
+    {components, [], subflow} = Manipulations.insert(components, index, new_components, name)
     composite = %{composite | components: components}
     {:reply, {composite, subflow}, composite}
   end

lib/composite/manipulations.ex

@@ -5,52 +5,30 @@ defmodule Strom.Composite.Manipulations do
   alias Strom.Composite.StartStop
   alias Strom.GenMix
 
-  def delete(components, index_from, index_to) do
-    {{new_components, deleted_components}, _} =
-      Enum.reduce(components, {{[], []}, 0}, fn component, {{acc, deleted_acc}, index} ->
-        cond do
-          index == index_from ->
-            next_component = Enum.at(components, index_to + 1)
-            input_streams = Strom.GenMix.state(component.pid).input_streams
-
-            {:ok, _gm_pid, new_tasks} =
-              GenServer.call(next_component.pid, {:start_tasks, input_streams})
-
-            GenServer.cast(component.pid, {:transfer_tasks, new_tasks})
-            {{acc, [component | deleted_acc]}, index + 1}
-
-          index > index_from and index <= index_to ->
-            GenServer.cast(component.pid, {:gen_mix, :stopping})
-            {{acc, [component | deleted_acc]}, index + 1}
-
-          true ->
-            {{[component | acc], deleted_acc}, index + 1}
-        end
-      end)
-
-    {Enum.reverse(new_components), Enum.reverse(deleted_components)}
-  end
-
+  @spec insert(list(Strom.component()), integer(), list(Strom.component()), atom()) ::
+          {list(Strom.component()), list(Strom.component()), Strom.flow()}
   def insert(components, index, new_components, name) when is_list(new_components) do
     component_after = Enum.at(components, index)
     gm_after = GenMix.state(component_after.pid)
 
     new_components = StartStop.start_components(new_components, name)
     flow = Composite.call_flow(new_components, gm_after.input_streams)
 
-    {:ok, _gm_pid, new_tasks} =
-      GenServer.call(component_after.pid, {:start_tasks, Map.take(flow, component_after.inputs)})
+    {_gm_pid, new_tasks} =
+      GenMix.start_tasks(component_after.pid, Map.take(flow, component_after.inputs))
 
-    GenServer.call(component_after.pid, {:replace_tasks, new_tasks})
+    GenMix.transfer_tasks(component_after.pid, new_tasks, :old)
 
     components =
       components
       |> List.insert_at(index, new_components)
       |> List.flatten()
 
-    {components, Map.drop(flow, component_after.inputs)}
+    {components, [], Map.drop(flow, component_after.inputs)}
   end
 
+  @spec replace(list(Strom.component()), integer(), integer(), list(Strom.component()), atom()) ::
+          {list(Strom.component()), list(Strom.component()), Strom.flow()}
   def replace(components, index_from, index_to, new_components, name)
       when is_list(new_components) do
     {{new_components, deleted_components, subflow}, _} =
@@ -61,16 +39,12 @@ defmodule Strom.Composite.Manipulations do
             input_streams = Strom.GenMix.state(component.pid).input_streams
             new_components = StartStop.start_components(new_components, name)
             flow = Composite.call_flow(new_components, input_streams)
-
             component_after = Enum.at(components, index_to + 1)
 
-            {:ok, _gm_pid, new_tasks} =
-              GenServer.call(
-                component_after.pid,
-                {:start_tasks, Map.take(flow, component_after.inputs)}
-              )
+            {_gm_pid, new_tasks} =
+              GenMix.start_tasks(component_after.pid, Map.take(flow, component_after.inputs))
 
-            GenServer.cast(component.pid, {:transfer_tasks, new_tasks})
+            GenMix.transfer_tasks(component.pid, new_tasks, :all)
 
             {{acc ++ Enum.reverse(new_components), [component | deleted_acc],
               Map.drop(flow, component_after.inputs)}, index + 1}

lib/gen_mix.ex

@@ -33,6 +33,9 @@ defmodule Strom.GenMix do
   alias Strom.GenMix.Streams
   alias Strom.GenMix.Tasks
 
+  @type t() :: %__MODULE__{}
+
+  @spec start(__MODULE__.t()) :: __MODULE__.t()
   def start(%__MODULE__{process_chunk: process_chunk, opts: opts, composite: composite} = gm)
       when is_list(opts) do
     gm = %{
@@ -69,20 +72,38 @@ defmodule Strom.GenMix do
   end
 
   @impl true
+  @spec init(__MODULE__.t()) :: {:ok, __MODULE__.t()}
   def init(%__MODULE__{} = gm) do
     {:ok, %{gm | pid: self()}}
   end
 
+  @spec start_tasks(pid(), Strom.flow()) :: {pid(), map()}
+  def start_tasks(gm_pid, input_streams) do
+    GenServer.call(gm_pid, {:start_tasks, input_streams})
+  end
+
+  @spec run_tasks(pid(), atom()) :: :ok
+  def run_tasks(gm_pid, output_name) do
+    GenServer.call(gm_pid, {:run_tasks, output_name})
+  end
+
   @spec call(map(), map()) :: map() | no_return()
   def call(flow, gm), do: Streams.call(flow, gm)
 
+  @spec state(pid()) :: __MODULE__.t()
   def state(pid), do: GenServer.call(pid, :state)
 
   @spec stop(any()) :: any()
   def stop(gm) do
     GenServer.call(gm.pid, :stop)
   end
 
+  @spec transfer_tasks(pid(), map(), atom()) :: :ok
+  def transfer_tasks(gm_pid, new_tasks, all_or_old) when all_or_old in [:all, :old] do
+    GenServer.cast(gm_pid, {:transfer_tasks, new_tasks, all_or_old})
+  end
+
+  @spec process_chunk(atom(), list(), Strom.flow(), any()) :: {Strom.flow(), boolean(), any()}
   def process_chunk(_input_stream_name, chunk, outputs, nil) do
     outputs
     |> Enum.reduce({%{}, false, nil}, fn {output_name, output_stream_fun}, {acc, any?, nil} ->
@@ -97,7 +118,7 @@ defmodule Strom.GenMix do
     new_tasks = Tasks.start_tasks(input_streams, gm)
     tasks = Map.merge(gm.tasks, new_tasks)
 
-    {:reply, {:ok, gm.pid, new_tasks},
+    {:reply, {gm.pid, new_tasks},
      %{gm | tasks_started: true, tasks_run: false, tasks: tasks, input_streams: input_streams}}
   end
 
@@ -120,16 +141,6 @@ defmodule Strom.GenMix do
     {:reply, :ok, %{gm | clients: clients}}
   end
 
-  def handle_call(
-        {:replace_tasks, new_tasks},
-        _from,
-        %__MODULE__{tasks_started: true} = gm
-      ) do
-    old_tasks = Map.drop(gm.tasks, Map.keys(new_tasks))
-    run_new_tasks_and_halt_the_old_ones(new_tasks, old_tasks, gm.accs)
-    {:reply, :ok, %{gm | tasks_run: true, tasks: new_tasks}}
-  end
-
   def handle_call(:stop, _from, %__MODULE__{} = gm) do
     gm = %{gm | stopping: true}
 
@@ -151,12 +162,19 @@ defmodule Strom.GenMix do
     after_action(%{gm | stopping: true})
   end
 
-  def handle_cast({:transfer_tasks, new_tasks}, gm) do
+  def handle_cast({:transfer_tasks, new_tasks, :all}, gm) do
     run_new_tasks_and_halt_the_old_ones(new_tasks, gm.tasks, gm.accs)
 
     after_action(%{gm | stopping: true})
   end
 
+  def handle_cast({:transfer_tasks, new_tasks, :old}, gm) do
+    old_tasks = Map.drop(gm.tasks, Map.keys(new_tasks))
+    run_new_tasks_and_halt_the_old_ones(new_tasks, old_tasks, gm.accs)
+
+    after_action(%{gm | tasks_run: true, tasks: new_tasks})
+  end
+
   def handle_cast(
         {:put_data, {input_name, task_pid}, {new_data, new_acc}},
         %__MODULE__{} = gm

lib/gen_mix/streams.ex

@@ -1,13 +1,15 @@
 defmodule Strom.GenMix.Streams do
   @moduledoc "Utility module. There are functions for manipulating data in gen_mix"
 
+  alias Strom.GenMix
+
   def call(flow, gm) do
     input_streams =
       Enum.reduce(gm.inputs, %{}, fn name, acc ->
         Map.put(acc, name, Map.fetch!(flow, name))
       end)
 
-    {:ok, gm_pid, _new_tasks} = GenServer.call(gm.pid, {:start_tasks, input_streams})
+    {gm_pid, _new_tasks} = GenMix.start_tasks(gm.pid, input_streams)
 
     sub_flow = build_sub_flow(gm.outputs, gm_pid)
 
@@ -21,7 +23,7 @@ defmodule Strom.GenMix.Streams do
       stream =
         Stream.resource(
           fn ->
-            :ok = GenServer.call(gm_pid, {:run_tasks, output_name})
+            :ok = GenMix.run_tasks(gm_pid, output_name)
             gm_pid
           end,
           fn gm_pid ->

lib/mixer.ex

@@ -54,6 +54,7 @@ defmodule Strom.Mixer do
     GenMix.call(flow, mixer)
   end
 
+  @spec process_chunk(atom(), list(), Strom.flow(), nil) :: {Strom.flow(), boolean(), nil}
   def process_chunk(_input_stream_name, chunk, outputs, nil) when map_size(outputs) == 1 do
     [stream_name] = Map.keys(outputs)
     {%{stream_name => chunk}, Enum.any?(chunk), nil}

lib/sink.ex

@@ -90,6 +90,7 @@ defmodule Strom.Sink do
     end)
   end
 
+  @spec process_chunk(atom(), list(), Strom.flow(), any()) :: {Strom.flow(), false, nil}
   def process_chunk(input_stream_name, _chunk, _outputs, nil) do
     {%{input_stream_name => []}, false, nil}
   end

lib/splitter.ex

@@ -70,6 +70,7 @@ defmodule Strom.Splitter do
     GenMix.call(flow, splitter)
   end
 
+  @spec process_chunk(atom(), list(), Strom.flow(), nil) :: {Strom.flow(), boolean(), nil}
   def process_chunk(_input_stream_name, chunk, outputs, nil) do
     outputs
     |> Enum.reduce({%{}, false, nil}, fn {stream_name, fun}, {acc, any?, nil} ->

lib/strom.ex

@@ -3,6 +3,7 @@ defmodule Strom do
 
   @type event() :: any()
   @type stream_name() :: any()
+  @type component() :: struct()
   @type stream() :: Enumerable.t(event())
   @type flow() :: %{optional(stream_name()) => stream()}
 end

lib/transformer.ex

@@ -88,6 +88,7 @@ defmodule Strom.Transformer do
     GenMix.call(flow, transformer)
   end
 
+  @spec process_chunk(atom(), list(), Strom.flow(), any()) :: {Strom.flow(), boolean(), any()}
   def process_chunk(input_stream_name, chunk, outputs, acc) do
     output_function = Map.get(outputs, input_stream_name)