Async Stream Colltaz Stream Elixir

async_stream.ex

defmodule Streamer do
  def async(enum) do
    Stream.resource(
      #Start
      fn ->
        origin = self
        spawn fn ->
          for x <- enum do
            send origin, {:elem, x}
          end
        end
      end,

      #Next
      fn pid -> 
        receive do
          {:elem, x} -> 
            {[x], pid}
        end
      end,

      #After
      fn pid -> pid end
    )
  end
end

defmodule Collatz do
  def stream(n) do
    n
    |> Stream.iterate(&collatz_next/1)
    |> Stream.take_while(&(&1 != 1))
    |> Stream.concat([1])
  end

  def next(n) when rem(n, 2) == 0, do: div(n, 2)
  def next(n), do: 3*n + 1
end

defmodule SimpleCounter do
  def start_link do
    Agent.start_link(fn -> %{} end)
  end

  def count(counter, item) do
    Agent.get_and_update(counter, 
      &Map.get_and_update(&1, item, fn 
        nil ->
          {1, 1}
        n -> 
          next = n + 1
          {next, next} 
      end)
    )
  end
end

{:ok, counter} = SimpleCounter.start_link

Streamer.stream(2345)
|> Stream.map(fn x -> 
  times = SimpleCounter.count(counter, :source)
  IO.puts "Source {value: #{x}, times: #{times}}"
  x 
end)
|> Streamer.async
|> Stream.map(&Integer.to_string/1)
|> Stream.map(fn x -> 
  times = SimpleCounter.count(counter, :to_string)
  IO.puts "To String: {value: #{x}, times: #{times}}"
  x 
end)
|> Streamer.async
|> Stream.map(&String.length/1)
|> Stream.map(fn x -> 
  times = SimpleCounter.count(counter, :to_string)
  IO.puts "Length: {value: #{x}, times: #{times}}"
  x 
end)
|> Enum.take(5)
|> IO.inspect

Agent.stop(counter)