defmodule CollectableUtils do @doc """ Routes an Enumerable into Collectables based on a `key_fun`. This function is essentially a cross between `Enum.into/2` and `Enum.group_by/2`. Rather than the simple lists of `Enum.group_by/2`, `into_by/3` populates a set of Collectables. Like `Enum.group_by/2`, the caller supplies a routing function `key_fun`, specifying a "bucket name" for each element to be placed into. Unlike `Enum.group_by/2`, when a bucket does not already exist, `initial_fun` is called with the new `key`, and must return a command, one of the following: * `{:new, Collectable.t()}` – creates a new bucket with the given Collectable as the initial value * `:discard` – drops the element and continues processing * `:invalid` – stops processing and returns an error `{:invalid, element, key}` For convenience, rather than an `initial_fun`, you may pass a tuple `{initial_map, not_found_command}`. Keys existing in `initial_map` will translate to `{:new, value}`. Other keys will return `not_found_command`. The default behavior if `initial_fun` is not passed, is to imitate the behavior of `Enum.group_by/2`. This is equivalent to passing `{%{}, {:new, []}}` as an `initial_fun`. ## Examples # imitate Enum.group_by/3 more explicitly iex> CollectableUtils.into_by!( ...> 0..5, ...> fn i -> :erlang.rem(i, 3) end, ...> fn _ -> {:new, []} end ...> ) %{0 => [0, 3], 1 => [1, 4], 2 => [2, 5]} # Enum.group_by/3, but with MapSets, and for only existing keys iex> CollectableUtils.into_by!( ...> 0..5, ...> fn i -> :erlang.rem(i, 3) end, ...> {Map.new([0, 1], &{&1, MapSet.new()}), :discard} ...> ) %{ 0 => %MapSet{map: %{0 => [], 3 => []}}, 1 => %MapSet{map: %{1 => [], 4 => []}} } ## Recipes # Distribute lines to files by hash CollectableUtils.into_by( event_stream, fn event_ln -> :erlang.phash2(event_ln) |> :erlang.rem(256) end, fn shard -> {:new, File.stream!("events-\#{shard}.events")} end ) # Hourly log rotation CollectableUtils.into_by( log_stream, fn _ -> :erlang.system_time(:second) |> :erlang.div(3600) end, fn period -> {:new, File.stream!("app-\#{period}.log")} end ) """ def into_by(enumerable, key_fun, initial_fun \\ {%{}, {:new, []}}) do initial_fun = case initial_fun do f when is_function(f) -> f {initial_map, not_found_command} when is_map(initial_map) -> fn k -> default_initial_fun(initial_map, k, not_found_command) end end try do populated_collectors = Enum.reduce(enumerable, %{}, fn element, collectors -> key = key_fun.(element) case Map.fetch(collectors, key) do {:ok, {original, collector_fun}} -> Map.put(collectors, key, {collector_fun.(original, {:cont, element}), collector_fun}) :error -> case initial_fun.(key) do {:new, new_collectable} -> {new_original, collector_fun} = Collectable.into(new_collectable) Map.put_new( collectors, key, {collector_fun.(new_original, {:cont, element}), collector_fun} ) :discard -> collectors :invalid -> throw {:invalid, element, key} end end end) final_collectables = Map.new(populated_collectors, fn {name, {updated, collector_fun}} -> {name, collector_fun.(updated, :done)} end) {:ok, final_collectables} catch {:invalid, element, key} -> {:invalid, element, key} end end @doc """ Routes an Enumerable into Collectables based on a `key_fun`, erroring out if if a Collectable cannot be constructed for a key. Like `into_by/3`, but raises an ArgumentError if `initial_fun` returns `:invalid`. """ def into_by!(enumerable, key_fun, initial_fun \\ {%{}, {:new, []}}) do case into_by(enumerable, key_fun, initial_fun) do {:ok, final_collectables} -> final_collectables {:invalid, element, key} -> raise ArgumentError, "invalid key #{inspect key} derived for element: #{inspect element}" end end defp default_initial_fun(initial_map, key, not_found_command) do case Map.fetch(initial_map, key) do {:ok, value} -> {:new, value} :error -> not_found_command end end end