Stream.mutate/3

mutate.exs

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

Example: pass through values

def passthrough(enum) do
  MyStream.mutate enum, [], fn
    {:cont, val}, acc ->
      {[val], acc}
    :halt, _acc ->
      :halted
  end
end

iex> 1..5 |> passthrough |> Enum.to_list
[1, 2, 3, 4, 5]

Example: duplicate values from input stream

def duplicate(enum) do
  MyStream.mutate enum, [], fn
    {:cont, val}, acc ->
      {[val, val], acc}
    :halt, _acc ->
      :halted
  end
end

iex> 1..5 |> duplicate |> Enum.to_list
[1, 1, 2, 2, 3, 3, 4, 4, 5, 5]

Example: lookahead

def lookahead(enum, n) do
  mutate enum, {:buffer, []}, fn
    {:cont, val}, {:buffer, buffer} when length(buffer) < n ->
      {[], {:buffer, buffer ++ [val]}}
    {:cont, val}, {:buffer, buffer} ->
      {[], {:emit, buffer ++ [val]}}
    {:cont, val}, {:emit, [_|rest] = buffer} ->
      {[buffer], {:emit, rest ++ [val]}}
    :halt, {_, [_|rest] = buffer} ->
      {[buffer], {:emit, rest}}
    :halt, _ ->
      :halted
  end
end

iex> 1..5 |> lookahead(2) |> Enum.to_list
[[1, 2, 3], [2, 3, 4], [3, 4, 5], [4, 5], [5]]

Example: lookbehind

def lookbehind(enum, n) do
  MyStream.mutate enum, {:buffer, []}, fn
    {:cont, val}, {:buffer, []} ->
      {[], {:buffer, [val]}}
    {:cont, val}, {:buffer, buffer} when length(buffer) < n ->
      {[buffer], {:buffer, buffer ++ [val]}}
    {:cont, val}, {:buffer, buffer} ->
      {[buffer], {:emit, buffer ++ [val]}}
    {:cont, val}, {:emit, [_|rest] = buffer} ->
      {[buffer], {:emit, rest ++ [val]}}
    :halt, {_, []} ->
      :halted
    :halt, {_, buffer} ->
      {[buffer], {:halt, []}}
  end
end

iex> 1..5 |> lookbehind(2) |> Enum.to_list
[[1], [1, 2], [1, 2, 3], [2, 3, 4], [3, 4, 5]]