concurrent

defmodule Concurrent do
  @moduledoc """
  Lightweight Enum with concurrency
  """

  @doc """
  Concurrently iterates over an Enum

  ## Examples

      iex> Concurrent.each([1, 2, 3], &IO.inspect(&1))
      :ok

  """
  def each(enum, fun, options \\ []) do
    opts = Enum.into(options, %{timeout: 5_000, batch_size: 10})

    enum
    |> Enum.chunk_every(opts.batch_size)
    |> Enum.each(fn batch ->
      batch
      |> Enum.map(&Task.async(fn -> fun.(&1) end))
      |> Enum.each(&Task.await(&1, opts.timeout))
    end)
  end

  @doc """
  Concurrently maps over an Enum

  ## Examples

      iex> Concurrent.map([1, 2, 3], &(&1 * &1))
      [1, 4, 9]

  """
  def map(enum, fun, options \\ []) do
    opts = Enum.into(options, %{timeout: 5_000, batch_size: 10})

    enum
    |> Enum.chunk_every(opts.batch_size)
    |> Enum.flat_map(fn batch ->
      batch
      |> Enum.map(&Task.async(fn -> fun.(&1) end))
      |> Enum.map(&Task.await(&1, opts.timeout))
    end)
  end
end