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
)

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.