algorithm.ex

defmodule Algorithm do
  # Helpers
  def parallel(collection, fun) when is_function(fun) do
    master = self()

    collection
    |> Enum.map(fn item ->
      Task.start fn ->
        fun.(item)
        send(master, self())
      end
    end)
    |> Enum.each(fn {:ok, pid} ->
      receive do
        ^pid -> :ok
      after
        timeout -> :error
      end
    end)
  end

  # Walk a collection
  def map(enum, fun) when is_list(enum) and is_function(fun) do
    enum
    |> Enum.chunk(chunk_size, chunk_size, [])
    |> Enum.each(&parallel(&1, fun))
  end

  def map(stream, fun) when is_function(fun) do
    stream
    |> Stream.chunk(chunk_size, chunk_size, [])
    |> Stream.each(&parallel(&1, fun))
  end

  # Config
  def chunk_size do
    Application.get_env(:algorithm, :request_chunk_size)
  end

  def timeout do
    Application.get_env(:algorithm, :request_timeout)
  end
end