algorithm.ex

defmodule Algorithm do
  require Logger

  # Helpers
  def pmap(collection, fun) when is_function(fun) do
    master = self()

    collection
    |> Enum.map(fn elem ->
      result_pid = spawn_link fn ->
        receive do
          {_pid, result} -> send(master, {self(), result})
        end
      end

      spawn_link fn ->
        send(result_pid, {self(), fun.(elem)})
      end

      result_pid
    end)
    |> Enum.flat_map(fn pid ->
      receive do
        {^pid, result} -> [result]
      after
        timeout -> []
      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.flat_map(&pmap(&1, fun))
  end

  def map(stream, fun) when is_function(fun) do
    stream
    |> Stream.chunk(chunk_size, chunk_size, [])
    |> Stream.flat_map(&pmap(&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