transducer

# Copyright 2016 Gary Poster
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

defprotocol Transducer do

  @type accumulator :: any
  @type state :: any
  @type annotated_accumulator :: {:cont, accumulator} | {:halt, accumulator} | {:reduce, term, accumulator}
  @type stateless_reducer :: (term, accumulator -> annotated_accumulator)
  @type stateful_reducer :: (term, {state, accumulator} -> annotated_accumulator)
  @type reducer_function :: stateless_reducer | stateful_reducer

  @spec reducer(any, reducer_function) :: reducer_function
  def reducer(transducer, reducer)

  @spec initial_state(any) :: any
  def initial_state(transducer)

  @spec compose(any, any) :: any
  def compose(transducer, other)
end

defmodule StatefulTransducer do
  defstruct initial_state: nil, function: nil
end

defmodule ComposedTransducer do
  defstruct transducers: []
end

defimpl Transducer, for: Function do
  def reducer(transducer, reducer_function), do: transducer.(reducer_function)
  def initial_state(_), do: :stateless
  def compose(transducer, %StatefulTransducer{}=other) do
    %ComposedTransducer{transducers: [transducer, other]}
  end
  def compose(transducer, %ComposedTransducer{}=other) do
    %ComposedTransducer{transducers: [transducer | other.transducers]}
  end
  def compose(transducer, other), do: fn reducer -> transducer.(other.(reducer)) end
end

defimpl Transducer, for: StatefulTransducer do
  def reducer(transducer, reducer_function) do
    transducer.function.(reducer_function)
  end
  def initial_state(transducer), do: transducer.initial_state
  def compose(transducer, %ComposedTransducer{}=other) do
    %ComposedTransducer{transducers: [transducer | other.transducers]}
  end
  def compose(transducer, other) do
    %ComposedTransducer{transducers: [transducer, other]}
  end
end

defimpl Transducer, for: ComposedTransducer do
  defp short_circuit(transducer) do
    Transducer.reducer(transducer, fn item, acc -> {:reduce, item, acc} end)
  end
  def reducer(transducer, final_reducer) do
    reducers = Enum.map(transducer.transducers, &short_circuit/1) ++ [final_reducer]
    fn item, {states, accumulator} ->
      reduce_composed(item, accumulator, reducers, states, [])
    end
  end
  defp reduce_composed(item, accumulator, [reducer | reducers], [:stateless | states], used_states) do
    case reducer.(item, accumulator) do
      {:reduce, item, accumulator} ->
        reduce_composed(item, accumulator, reducers, states, [:stateless | used_states])
      {:halt, accumulator} ->
        {:halt, {Enum.reverse([:stateless | used_states]) ++ states, accumulator}}
      {:cont, accumulator} ->
        {:cont, {Enum.reverse([:stateless | used_states]) ++ states, accumulator}}
    end
  end
  defp reduce_composed(item, accumulator, [reducer | reducers], [state | states], used_states) do
    case reducer.(item, {state, accumulator}) do
      {:reduce, item, {state, accumulator}} ->
        reduce_composed(item, accumulator, reducers, states, [state | used_states])
      {:halt, {state, accumulator}} ->
        {:halt, {Enum.reverse([state | used_states]) ++ states, accumulator}}
      {:cont, {state, accumulator}} ->
        {:cont, {Enum.reverse([state | used_states]) ++ states, accumulator}}
    end
  end
  defp reduce_composed(item, accumulator, [reducer], [], used_states) do
    reducer.(item, {Enum.reverse(used_states), accumulator})
  end
  def initial_state(transducer) do
    Enum.map(transducer.transducers, &Transducer.initial_state/1)
  end
  def compose(transducer, %ComposedTransducer{transducers: [other | transducers]}) do
    _compose(transducer.transducers, other, [], transducers)
  end
  def compose(transducer, other) do
    _compose(transducer.transducers, other, [], [])
  end
  defp _compose([], nil, acc, [head | tail]) do
    _compose([], nil, [head | acc], tail)
  end
  defp _compose([], nil, acc, []) do
    %ComposedTransducer{transducers: Enum.reverse(acc)}
  end
  defp _compose([next], other, acc, tail) when is_function(next) and is_function(other) do
    _compose([], nil, [Transducer.compose(next, other) | acc], tail)
  end
  defp _compose([next], other, acc, tail) do
    _compose([], nil, [other | [next | acc]], tail)
  end
  defp _compose([next | transducers], other, acc, tail) do
    _compose(transducers, other, [next | acc], tail)
  end
end

defmodule Transduce do

  @moduledoc """
  Composable algorithmic transformations.

  Rich Hickey introduced the idea in [this
  post](http://blog.cognitect.com/blog/2014/8/6/transducers-are-coming).
  [This post](http://phuu.net/2014/08/31/csp-and-transducers.html) is a good
  conceptual introduction.

  The input is always an enumerable.  The output can be an enumerable...

      iex> import Transduce, only: [transduce: 2, filter: 1, take: 1]
      iex> transduce(1..100, [filter(&(&1 > 5)), take(5)])
      [6, 7, 8, 9, 10]

...or can also produce another structure:

      iex> import Transduce, only: [transduce: 3, filter: 1, take: 1, put: 3]
      iex> transduce(
      ...>   [4, 8, 7, 3, 2, 9, 6, 12, 15], [
      ...>     filter(&(&1 > 5)),
      ...>     take(5),
      ...>     put(:min, nil, &min/2),
      ...>     put(:max, 0, &max/2),
      ...>     put(:count, 0, fn _, a -> a+1 end),
      ...>     put(:total, 0, &Kernel.+/2)],
      ...>   %{})
      %{count: 5, max: 12, min: 6, total: 42}

  You can write two kinds of transducers: stateless and stateful.  A stateless
  transducer is the most straightforward, since it is just a function.  Consider
  the `map` transducer.

  ```
  def map(f) do
    fn rf ->
      fn item, accumulator -> rf.(f.(item), accumulator) end
    end
  end
  ```

  At its simplest, a transducer takes a reducing function (`rf` above), and
  then returns another reducing function that wraps the `rf` with its own
  behavior.  For `map`, it passes the mapped value to the `rf` for the next
  step.  This works if the rf eventually returns a value.

  The return value is annotated, as defined by the Enumerable protocol.
  `{:cont, VALUE}` specifies that the reduction process should continue with
  the next item in the enumerable input, but VALUE is the new accumulator.
  `{:halt, VALUE}` specifies that the reduction process should stop,
  short-circuiting, and return VALUE as the result.

  For example, consider the `filter` implementation.

  ```
  def filter(f) do
    fn rf ->
      fn item, accumulator ->
        if f.(item) do rf.(item, accumulator) else {:cont, accumulator} end
      end
    end
  end
  ```

  In that case, if the filter passes the new item from the enumerable, it is
  passed through to the inner reducing function.  If it doesn't pass, the
  code returns `:cont` with an unchanged accumulator, indicating that we should
  move on to the next item in the source enumerable without a new accumulator.
  For an example of `:halt`, see the `take_while` implementation.

  A stateful transducer looks similar, but has some additional features.
  Consider the `take` implementation.

  ```
  def take(count) do
    %StatefulTransducer{
      initial_state: 0,
      function: fn rf ->
        fn
          item, {state, accumulator} when state < count ->
            rf.(item, {state+1, accumulator})
          item, {state, accumulator} -> {:halt, {state, accumulator}}
        end
      end
    }
  end
  ```

  A take transducer returns a StatefulTransducer struct, which specifies an
  `initial_state` and a `function`.  The function again takes a reducing
  function (`rf`) but the wrapping function this time expects that the
  accumulator has the shape `{state, accumulator}`.  The state is private to
  this function, and will not be in the final accumulator result, but must also
  be included in the function output, whether it's passed to the wrapped
  reducing function or returned to the caller with `:halt` or `:cont`.
  """

  @doc ~S"""
  Transduce a given enumerable to generate a list.

  ## Examples

      iex> import Transduce, only: [transduce: 2, take: 1, compose: 1, filter: 1]
      iex> transduce([2,3,5,7,11], take(3))
      [2, 3, 5]
      iex> transduce(0..20, compose([filter(&(rem(&1, 2) == 0)), take(5)]))
      [0, 2, 4, 6, 8]
      iex> transduce(0..20, [filter(&(rem(&1, 3) == 0)), take(6)])
      [0, 3, 6, 9, 12, 15]
  """
  def transduce(enumerable, transducer) do
    transduce(enumerable, transducer, [], &({:cont, [&1 | &2]}), &:lists.reverse/1)
  end

  @spec transduce(any, any, any, Transducer.stateless_reducer, any) :: any
  def transduce(enumerable, transducer, accumulator, stateless_reducer \\ fn _, acc -> {:cont, acc} end, finalizer \\ &(&1))
  def transduce(enumerable, transducer, accumulator, stateless_reducer, finalizer) when is_function(transducer) do
    {_, result} = Enumerable.reduce(
      enumerable, {:cont, accumulator}, transducer.(stateless_reducer))
    finalizer.(result)
  end
  def transduce(enumerable, transducers, accumulator, stateless_reducer, finalizer) when is_list(transducers) do
    transduce(enumerable, compose(transducers), accumulator, stateless_reducer, finalizer)
  end

  def transduce(enumerable, transducer, accumulator, stateless_reducer, finalizer) do
    final_reducer = fn element, {state, accumulator} ->
       {atom, accumulator} = stateless_reducer.(element, accumulator)
       {atom, {state, accumulator}}
    end
    {_, {_state, result}} = Enumerable.reduce(
      enumerable,
      {:cont, {Transducer.initial_state(transducer), accumulator}},
      Transducer.reducer(transducer, final_reducer))
    finalizer.(result)
  end

  @doc ~S"""
  Compose multiple transducers into one.

  ## Examples

      iex> import Transduce, only: [transduce: 2, compose: 1, map: 1, filter: 1]
      iex> transduce([2,3,5,7,11,13,17,19,23], compose([map(&(&1+1)), filter(&(rem(&1,3)==0))]))
      [3, 6, 12, 18, 24]

  Stateless transducers compose into functions.

      iex> import Transduce, only: [compose: 1, map: 1]
      iex> tr = compose([map(&(&1*2)), map(&(&1+2))])
      iex> tr.(fn item, acc -> {item, acc} end).(5, 42)
      {12, 42}

  If any other kind of transducer enters the mix, it becomes a
  ComposedTransducer.

      iex> import Transduce, only: [compose: 1, map: 1, take: 1]
      iex> tr = compose([map(&(&1*2)), take(5)])
      iex> length(tr.transducers)
      2

  Composed transducers can themselves be composed.

      iex> import Transduce, only: [transduce: 2, compose: 1, map: 1, filter: 1, take: 1]
      iex> tr1 = compose([filter(&(rem(&1, 3)==0)), map(&(&1*2))])
      iex> tr2 = compose([map(&(&1+1)), take(5)])
      iex> transduce(0..20, compose([tr1, tr2]))
      [1, 7, 13, 19, 25]
  """
  def compose([first | rest] = _transducers) do
    _compose(first, rest)
  end

  def _compose(current, [next1, next2 | rest])
    when not is_function(current) and is_function(next1) and is_function(next2) do
    _compose(current, [Transducer.compose(next1, next2) | rest])
  end
  def _compose(current, [next | rest]) do
    _compose(Transducer.compose(current, next), rest)
  end
  def _compose(current, []), do: current

  @doc ~S"""
  Apply a function to each item it receives.

  ## Examples

      iex> import Transduce, only: [transduce: 2, map: 1]
      iex> transduce(0..4, map(&(-&1)))
      [0, -1, -2, -3, -4]
  """
  def map(f) do
    fn rf ->
      fn item, accumulator -> rf.(f.(item), accumulator) end
    end
  end

  @doc ~S"""
  Only include items if the filter function returns true.

  ## Examples

      iex> import Transduce, only: [transduce: 2, filter: 1]
      iex> transduce(0..5, filter(&(rem(&1,2)==0)))
      [0, 2, 4]
  """
  def filter(f) do
    fn rf ->
      fn item, accumulator ->
        if f.(item) do rf.(item, accumulator) else {:cont, accumulator} end
      end
    end
  end

  @doc ~S"""
  Exclude items if the remove function returns true.

  ## Examples

      iex> import Transduce, only: [transduce: 2, remove: 1]
      iex> transduce(0..5, remove(&(rem(&1,2)==0)))
      [1, 3, 5]
  """
  def remove(f) do
    fn rf ->
      fn item, accumulator ->
        if f.(item) do {:cont, accumulator} else rf.(item, accumulator) end
      end
    end
  end

  @doc ~S"""
  Only iterate while the function returns true.

  ## Examples

      iex> import Transduce, only: [transduce: 2, take_while: 1]
      iex> transduce([0, 1, 2, 10, 11, 4], take_while(&(&1 < 10)))
      [0, 1, 2]
  """
  def take_while(f) do
    fn rf ->
      fn item, accumulator ->
        if f.(item) do rf.(item, accumulator) else {:halt, accumulator} end
      end
    end
  end

  @doc ~S"""
  Take the first N items from the enumerable and then stop iteration.

  ## Examples

      iex> import Transduce, only: [transduce: 2, take: 1, filter: 1]
      iex> transduce(0..200, take(5))
      [0, 1, 2, 3, 4]
      iex> transduce(0..200, [filter(&(rem(&1, 5)==0)), take(5)])
      [0, 5, 10, 15, 20]
  """
  def take(count) do
    %StatefulTransducer{
      initial_state: 0,
      function: fn rf ->
        fn
          item, {state, accumulator} when state < count ->
            rf.(item, {state+1, accumulator})
          _, {state, accumulator} -> {:halt, {state, accumulator}}
        end
      end
    }
  end

  @doc ~S"""
  Skip the first N items from the enumerable and then iterate.

  ## Examples

      iex> import Transduce, only: [transduce: 2, skip: 1, take: 1]
      iex> transduce(0..10, skip(8))
      [8, 9, 10]
      iex> transduce(0..20, [skip(4), take(2)])
      [4, 5]
  """
  def skip(count) do
    %StatefulTransducer{
      initial_state: 0,
      function: fn rf ->
        fn
          _, {state, accumulator} when state < count ->
            {:cont, {state + 1, accumulator}}
          item, {state, accumulator} -> rf.(item, {state + 1, accumulator})
        end
      end
    }
  end

  @doc ~S"""
  Call the function with each value and the result of the previous call,
  beginning with the initial_value.

  ## Examples

      iex> import Transduce, only: [transduce: 2, scan: 2]
      iex> transduce(1..5, scan(0, &(&1 + &2)))
      [1, 3, 6, 10, 15]
  """
  def scan(initial_value, f) do
    %StatefulTransducer{
      initial_state: initial_value,
      function: fn rf ->
        fn item, {state, accumulator} ->
          new = f.(item, state)
          rf.(new, {new, accumulator})
        end
      end
    }
  end

  @doc ~S"""
  Step over N items in the enumerable, taking 1 between each set.  Called with
  two arguments, you specify how many to take (take_count, skip_count).

  ## Examples

      iex> import Transduce, only: [transduce: 2, step: 1, step: 2]
      iex> transduce(0..10, step(2))
      [0, 3, 6, 9]
      iex> transduce(0..15, step(2, 3))
      [0, 1, 5, 6, 10, 11, 15]
  """
  def step(skip_count), do: step(1, skip_count)
  def step(take_count, skip_count) do
    total = take_count + skip_count
    %StatefulTransducer{
      initial_state: 0,
      function: fn rf ->
        fn item, {state, accumulator} ->
          position = rem(state, total)
          if position < take_count do
            rf.(item, {position+1, accumulator})
          else
            {:cont, {position+1, accumulator}}
          end
        end
      end
    }
  end

  @doc ~S"""
  For each item in the enumerable, transform it and the previous result with the
  given reducer and then stash the resulting value in the accumulator, which
  must be a map.

  ## Examples

      iex> import Transduce, only: [transduce: 3, put: 3]
      iex> transduce([6,3,8,2,4,9,5,0,1,7], [put(:min, nil, &min/2), put(:max, 0, &max/2)], %{})
      %{max: 9, min: 0}

      iex> import Transduce, only: [transduce: 3, filter: 1, put: 3]
      iex> transduce(
      ...>   1..20, [
      ...>     put(:total, 0, &Kernel.+/2),
      ...>     put(:count, 0, fn _, ct -> ct + 1 end),
      ...>     filter(fn v -> rem(v, 2) == 0 end),
      ...>     put(:even, 0, &Kernel.+/2)
      ...>   ],
      ...>   %{})
      %{count: 20, even: 110, total: 210}

      iex> import Transduce, only: [transduce: 3, filter: 1, put: 4]
      iex> transduce(
      ...>   1..20, [
      ...>     put(:even, 0, filter(&(rem(&1, 2) == 0)), &Kernel.+/2),
      ...>     put(:odd, 0, filter(&(rem(&1, 2) == 1)), &Kernel.+/2)
      ...>   ],
      ...>   %{})
      %{even: 110, odd: 100}

      iex> import Transduce, only: [transduce: 3, take: 1, put: 4, put: 3]
      iex> transduce(
      ...>   1..20, [
      ...>     put(:first, 0, take(5), &Kernel.+/2),
      ...>     put(:total, 0, &Kernel.+/2)
      ...>   ],
      ...>   %{})
      %{first: 15, total: 210}
  """
  def put(key, initial_value, reducer) do
    fn rf ->
      fn item, acc ->
        rf.(item, Map.put(acc, key, reducer.(item, Map.get(acc, key, initial_value))))
      end
    end
  end
  def put(key, initial_value, transducers, reducer) when is_list(transducers) do
    put(key, initial_value, compose(transducers), reducer)
  end
  def put(key, initial_value, transducer, reducer) when is_function(transducer) do
    final_reducer = Transducer.reducer(
      transducer,
      fn item, accumulator ->
        { :cont,
          Map.put(
            accumulator,
            key,
            reducer.(item, Map.get(accumulator, key, initial_value)))
        }
      end
    )
    %StatefulTransducer{
      initial_state: :cont,
      function: fn rf ->
        fn
          item, {:halt, _} = accumulator -> rf.(item, accumulator)
          item, {:cont, accumulator} -> rf.(item, final_reducer.(item, accumulator))
        end
      end
    }
  end
  def put(key, initial_value, transducer, reducer) do
    final_reducer = Transducer.reducer(
      transducer,
      fn item, {state, accumulator} ->
        { :cont,
          { state,
            Map.put(
              accumulator,
              key,
              reducer.(item, Map.get(accumulator, key, initial_value)))
          }
        }
      end)
    %StatefulTransducer{
      initial_state: {:cont, Transducer.initial_state(transducer)},
      function: fn rf ->
        fn
          item, {{:halt, _}, _}=accumulator -> rf.(item, accumulator)
          item, {{:cont, state}, accumulator} ->
            {disposition, {new_state, new_accumulator}} =
              final_reducer.(item, {state, accumulator})
            rf.(item, {{disposition, new_state}, new_accumulator})
        end
      end
    }
  end
end