Provide Enum.chunk_by/4 and Stream.chunk_by/4

[josevalim]

This is a generalization of the current Stream.chunk_by/2 function.

Example:

1..10
|> Stream.chunk_by([], fn i, acc ->
  if rem(i, 2) == 0, do: {Enum.reverse([i | acc]), []}, else: {[], [i | acc]}
end, fn acc -> Enum.reverse(acc) end)
|> Enum.to_list
#=> [[1, 2], [3, 4], [5, 6], [7, 8], ...]

[iagooar]

Hello José,

I'd love to contribute by adding this feature. This is my very first time contributing to Elixir, so I'd need some guidance here.

So far I've been diving into the source code and got stuck with some open questions. Hope you can help me understand better how stuff works here.

I started by adding the function chunk_by/4 (the body of the function is a copy of chunk_by/2 with minor changes, in order to have a starting point):

  @doc """
  Chunks the `enum` by buffering elements for which `fun` returns
  the same value and only emit them when `fun` returns a new value
  or the `enum` finishes.

  ## Examples

  iex> start_func = fn i, acc -> if rem(i, 2) == 0, do: {Enum.reverse([i | acc]), []}, else: {[], [i | acc]} end
  iex> stream = Stream.chunk_by(1..10, [], start_func, fn acc -> Enum.reverse(acc) end)
  iex> Enum.to_list(stream)
  [[1, 2], [3, 4], [5, 6], [7, 8], [9, 10]]
  """

  @spec chunk_by(Enumerable.t, Enumerable.t, (element, acc -> {Enumerable.t, acc}), (acc -> Enumerable.t)) :: Enumerable.t
  def chunk_by(enum, acc, start_fun, after_fun) do
    lazy enum, nil,
         fn(f1) -> R.chunk_by(start_fun, f1) end,
         &do_chunk_by(&1, &2)
  end

Then I added this basic test:

  test "chunk_by/4" do
    stream = Stream.chunk_by(1..10, [], fn i, acc ->
      if rem(i, 2) == 0, do: {Enum.reverse([i | acc]), []}, else: {[], [i | acc]}
    end, fn acc -> Enum.reverse(acc) end)

    assert lazy?(stream)
    assert Enum.to_list(stream) ==
      [[1, 2], [3, 4], [5, 6], [7, 8], [9, 10]]
  end

After running the test, I found that I have to change the reducer macro chunk_by/2 in stream/reducers.ex because the arity of start_fun is different from the function in Stream.chunk_by/2.

  defmacro chunk_by(callback, fun \\ nil) do
    quote do
      fn(entry, acc(head, {buffer, value}, tail) = acc) ->
        new_value = unquote(callback).(entry, acc)
        if new_value == value do
          skip(acc(head, {[entry | buffer], value}, tail))
        else
          next_with_acc(unquote(fun), :lists.reverse(buffer), head, {[entry], new_value}, tail)
        end
        entry, acc(head, nil, tail) = acc ->
          skip(acc(head, {[entry], unquote(callback).(entry, acc)}, tail))
      end
    end
  end

When running the tests, I get the following failure:

  5) test chunk_by/4 (StreamTest)
     lib/elixir/test/elixir/stream_test.exs:99
     Assertion with == failed
     code:  Enum.to_list(stream) == [[1, 2], [3, 4], [5, 6], [7, 8], '\t\n']
     left:  [[1], [2], [3], [4], [5], [6], '\a', '\b', '\t', '\n']
     right: [[1, 2], [3, 4], [5, 6], [7, 8], [9, 10]]
     stacktrace:
       lib/elixir/test/elixir/stream_test.exs:105: (test)

From here I have a bit of a struggle with knowing where to go next. Any input will be greatly appreciated :)

[josevalim]

@YAGoOaR can you please send a PR? This way I can check out your code and give more precise feedback. :) Also, please make sure to give me push permission to your branch.