defmodule Plymio.Funcio.Enum.Map do @moduledoc ~S""" Map Functions for an Enumerable. See `Plymio.Funcio` for overview and documentation terms. > The concurrent functions currently produce a `List` rather than a `Stream`. But this could change in the future. The examples below assumes a `Stream` was returned. """ use Plymio.Funcio.Attribute @type opts :: Plymio.Funcio.opts() @type error :: Plymio.Funcio.error() @type result :: Plymio.Funcio.result() @type fun1_map :: Plymio.Funcio.fun1_map() import Plymio.Fontais.Option, only: [ opts_normalise: 1, opts_validate: 1, opts_get: 3 ] import Plymio.Funcio.Enum.Collate, only: [ collate0_enum: 1 ] import Plymio.Funcio.Map.Utility, only: [ reduce_or_passthru_map1_funs: 1, reduce_map1_funs: 1 ] @doc ~S""" `map_enum/3` takes an *enum* and a *map/1*, reduces the *map/1* to a single, composite function and applied the composite to each value in the *enum* returning `{:ok, values}` where `values` *may* be a `Stream`. ## Examples iex> enum = 0 .. 4 ...> fun_map = fn v -> v * v end ...> {:ok, values} = enum ...> |> map_enum(fun_map) ...> values |> Enum.to_list [0, 1, 4, 9, 16] iex> enum = 0 .. 4 ...> fun_map1 = fn v -> v + 1 end ...> fun_map2 = fn v -> v * v end ...> fun_map3 = fn v -> v - 1 end ...> {:ok, values} = enum ...> |> map_enum([fun_map1, fun_map2, fun_map3]) ...> values |> Enum.to_list [0, 3, 8, 15, 24] iex> enum = 0 .. 99 |> Stream.map(&(&1)) ...> fun_map = fn v -> v end ...> {:ok, values} = enum ...> |> map_enum(fun_map) ...> values |> Enum.reduce(0, fn v,s -> s + v end) 4950 iex> enum = :not_an_enum ...> fun_map = fn v -> v * v end ...> {:ok, stream} = enum |> map_enum(fun_map) ...> try do ...> stream |> Enum.to_list ...> rescue ...> error -> error ...> end ...> |> Exception.message ...> |> String.starts_with?("protocol Enumerable not implemented for :not_an_enum") true """ @since "0.1.0" @spec map_enum(any, fun) :: result def map_enum(enum, fun) def map_enum(enum, fun) do with {:ok, fun_map} <- fun |> reduce_or_passthru_map1_funs do try do {:ok, enum |> Stream.map(fun_map)} rescue error -> {:error, error} end else {:error, %{__exception__: true}} = result -> result end end @doc ~S""" `map_concurrent_enum/3` takes an *enum* and a *map/1*, reduces the *map/1* to a single, composite function and applied the composite to each value in the *enum* in its own, separate task using using `Task.Supervisor.async_stream_nolink/4`. It returns `{:ok, values}` where the `values` *may* be a `Stream`. ## Examples iex> enum = 0 .. 4 ...> fun_map = fn v -> v * v end ...> {:ok, values} = enum ...> |> map_concurrent_enum(fun_map) ...> values |> Enum.to_list [0, 1, 4, 9, 16] iex> enum = 0 .. 4 ...> fun_map1 = fn v -> v + 1 end ...> fun_map2 = fn v -> v * v end ...> fun_map3 = fn v -> v - 1 end ...> {:ok, values} = enum ...> |> map_concurrent_enum([fun_map1, fun_map2, fun_map3]) ...> values |> Enum.to_list [0, 3, 8, 15, 24] iex> enum = 0 .. 99 ...> fun_map = fn v -> v end ...> {:ok, values} = enum |> map_concurrent_enum(fun_map) ...> values |> Enum.reduce(0, fn v, s -> s + v end) 4950 iex> enum = :not_an_enum ...> fun_map = fn v -> v * v end ...> {:error, error} = enum |> map_concurrent_enum(fun_map) ...> error |> Exception.message ...> |> String.starts_with?("protocol Enumerable not implemented for :not_an_enum") true """ @since "0.1.0" @spec map_concurrent_enum(any, fun, opts) :: result def map_concurrent_enum(enum, fun, opts \\ []) def map_concurrent_enum(enum, fun, opts) do with {:ok, opts} <- opts |> opts_normalise, {:ok, fun_map} <- fun |> reduce_or_passthru_map1_funs, {:ok, task_sup} <- opts |> opts_resolve_task_sup_pid, {:ok, async_stream_opts} <- opts |> opts_resolve_task_sup_async_stream_opts, true <- true do try do task_stream = task_sup |> Task.Supervisor.async_stream_nolink(enum, fun_map, async_stream_opts) with {:ok, _results} = result <- task_stream |> realise_task_stream_results, {:ok, _} <- task_sup |> stop_task_supervisor, true <- true do result else {:error, %{__exception__: true}} = result -> result end rescue error -> {:error, error} end else {:error, %{__exception__: true}} = result -> result end end @doc ~S""" `map_with_index_enum/2` take a *enum* and a *index map/1* and maps the *enum* with `Stream.with_index/1` and then `Stream.map/2` returning `{:ok, stream}`. The *index map/1* is called with `{value, index}` for each `value` in the *enum*. ## Examples iex> {:ok, stream} = [1,2,3] |> map_with_index_enum(fn {v,_i} -> v * v end) ...> stream |> Enum.to_list [1, 4, 9] iex> {:ok, stream} = [1,2,3] |> map_with_index_enum(fn {_v,i} -> i * i end) ...> stream |> Enum.to_list [0, 1, 4] iex> {:ok, stream} = [a: 1, b: 2, c: 3] ...> |> map_with_index_enum(fn {{_k,v},i} -> v * v * i end) ...> stream |> Enum.to_list [0, 4, 18] iex> {:ok, stream} = [a: 1, b: 2, c: 3] ...> |> map_with_index_enum([ ...> fn {{_k,v},i} -> {i, v * v} end, ...> fn {i, v} -> i + v end, ...> ]) ...> stream |> Enum.to_list [1, 5, 11] iex> enum = :not_an_enum ...> fun_map = fn v -> v * v end ...> {:ok, stream} = enum |> map_with_index_enum(fun_map) ...> try do ...> stream |> Enum.to_list ...> rescue ...> error -> error ...> end ...> |> Exception.message ...> |> String.starts_with?("protocol Enumerable not implemented for :not_an_enum") true iex> {:error, error} = [1,2,3] |> map_with_index_enum(:not_a_fun) ...> error |> Exception.message "map/1 function invalid, got: :not_a_fun" """ @since "0.1.0" @spec map_with_index_enum(any, any) :: {:ok, list} | {:error, error} def map_with_index_enum(derivable_list, mapper) def map_with_index_enum(state, mapper) do with {:ok, fun} <- mapper |> reduce_map1_funs, true <- true do try do {:ok, state |> Stream.with_index() |> Stream.map(fun)} rescue error -> {:error, error} end else {:error, %{__exception__: true}} = result -> result end end @doc ~S""" `map_concurrent_with_index_enum/2` take a *enum* and a *index map/1* and maps the *enum* with `Stream.with_index/1` and then `map_concurrent_enum/2` returning `{:ok, stream}`. The *index map/1* is called with `{value, index}` for each `value` in the *enum*. ## Examples iex> {:ok, stream} = [1,2,3] |> map_concurrent_with_index_enum(fn {v,_i} -> v * v end) ...> stream |> Enum.to_list [1, 4, 9] iex> {:ok, stream} = [1,2,3] |> map_concurrent_with_index_enum(fn {_v,i} -> i * i end) ...> stream |> Enum.to_list [0, 1, 4] iex> {:ok, stream} = [a: 1, b: 2, c: 3] ...> |> map_concurrent_with_index_enum(fn {{_k,v},i} -> v * v * i end) ...> stream |> Enum.to_list [0, 4, 18] iex> {:ok, stream} = [a: 1, b: 2, c: 3] ...> |> map_concurrent_with_index_enum([ ...> fn {{_k,v},i} -> {i, v * v} end, ...> fn {i, v} -> i + v end, ...> ]) ...> stream |> Enum.to_list [1, 5, 11] iex> {:error, error} = 42 |> map_concurrent_with_index_enum(&(&1)) ...> error |> Exception.message ...> |> String.starts_with?("protocol Enumerable not implemented for 42") true iex> {:error, error} = [1,2,3] |> map_concurrent_with_index_enum(:not_a_fun) ...> error |> Exception.message "map/1 function invalid, got: :not_a_fun" """ @since "0.1.0" @spec map_concurrent_with_index_enum(any, any) :: {:ok, list} | {:error, error} def map_concurrent_with_index_enum(enum, mapper) def map_concurrent_with_index_enum(state, mapper) do state |> Stream.with_index() |> map_concurrent_enum(mapper) end defp normalise_task_stream_result(value) defp normalise_task_stream_result({:ok, _} = result) do result end defp normalise_task_stream_result({:error, error}) do error |> normalise_task_stream_error_result! end defp normalise_task_stream_result({:exit, {value, _stacktrace}}) do value |> normalise_task_stream_error_result! end defp normalise_task_stream_results(results) do {:ok, results |> Stream.map(&normalise_task_stream_result/1)} end defp realise_task_stream_results(results) do with {:ok, stream} <- results |> normalise_task_stream_results do stream |> collate0_enum else {:error, %{__exception__: true}} = result -> result end end defp normalise_task_stream_error_value(value, exception) do cond do Exception.exception?(value) -> {:ok, value} true -> cond do Exception.exception?(exception) -> {:ok, exception} end end end defp normalise_task_stream_error_result(value, exception) do value |> normalise_task_stream_error_value(exception) |> case do {:ok, error} -> {:ok, {:error, error}} end end defp normalise_task_stream_error_result!(value, exception \\ nil) do value |> normalise_task_stream_error_result(exception) |> case do {:ok, result} -> result end end defp opts_resolve_task_sup_pid(opts) defp opts_resolve_task_sup_pid(opts) do with {:ok, opts} <- opts |> opts_validate, true <- true do opts |> Keyword.has_key?(@plymio_funcio_key_task_sup_pid) |> case do true -> opts |> Keyword.fetch(@plymio_funcio_key_task_sup_pid) _ -> with {:ok, sup_opts} <- opts |> opts_resolve_task_sup_start_link_opts, {:ok, _sup_pid} = result <- sup_opts |> Task.Supervisor.start_link(), true <- true do result else {:error, %{__exception__: true}} = result -> result end end else {:error, %{__exception__: true}} = result -> result end end defp opts_resolve_task_sup_start_link_opts(opts) defp opts_resolve_task_sup_start_link_opts( opts, defaults \\ @plymio_funcio_defaults_task_sup_start_link_opts ) do with {:ok, _sup_pid} = result <- opts |> opts_get(@plymio_funcio_key_task_sup_start_link_opts, defaults), true <- true do result else {:error, %{__exception__: true}} = result -> result end end defp opts_resolve_task_sup_async_stream_opts(opts) defp opts_resolve_task_sup_async_stream_opts( opts, defaults \\ @plymio_funcio_defaults_task_sup_async_stream_opts ) do with {:ok, _sup_pid} = result <- opts |> opts_get(@plymio_funcio_key_task_sup_async_stream_opts, defaults), true <- true do result else {:error, %{__exception__: true}} = result -> result end end defp stop_task_supervisor(pid) defp stop_task_supervisor(pid) when is_pid(pid) do pid |> Supervisor.stop() |> case do :ok -> {:ok, pid} end end end