defmodule MyStream do
  def mutate(enum, user_acc, user) do
    step = fn val, _acc -> {:suspend, val} end
    next = &Enumerable.reduce(enum, &1, step)
    &do_mutate([], user_acc, user, next, &1, &2)
  end

  defp do_mutate(values, user_acc, user, next, {:suspend, acc}, fun) do
    {:suspended, acc, &do_mutate(values, user_acc, user, next, &1, fun)}
  end

  defp do_mutate(_values, _user_acc, _user, _next, {:halt, acc}, _fun) do
    {:halted, acc}
  end

  defp do_mutate([], user_acc, user, next, {:cont, acc}, fun) do
    case next.({:cont, []}) do
      {:suspended, val, next} ->
        result = user.({:cont, val}, user_acc)
        mutate_user_result result, user, next, acc, fun
      {_, _} ->
        result = user.(:halt, user_acc)
        mutate_user_result result, user, next, acc, fun
    end
  end

  defp do_mutate([val|rest], user_acc, user, next, {:cont, acc}, fun) do
    do_mutate(rest, user_acc, user, next, fun.(val, acc), fun)
  end

  defp mutate_user_result({values, user_acc}, user, next, acc, fun) do
    do_mutate(values, user_acc, user, next, {:cont, acc}, fun)
  end

  defp mutate_user_result(:halted, _user, _next, acc, _fun) do
    {:halted, acc}
  end
end