           timed_out?: false,
         }
         running_tasks = Map.put(running_tasks, ref, task_info)
-        stream_monitor_loop(parent_pid, parent_ref, mfa, spawn, monitor_ref, running_tasks, timeout)
+        stream_monitor_loop(running_tasks, config)
 
       # The parent process is telling us to stop because the stream is being
       # closed. In this case, we forcely kill all spawned processes and then

    else
      {:suspended, [value], next} ->
        waiting = stream_spawn(value, spawned, waiting, monitor_pid, monitor_ref, timeout)
        stream_reduce({:cont, acc}, max - 1, spawned + 1, delivered, waiting, next,
                      reducer, monitor_pid, monitor_ref, timeout, on_timeout)
      {_, [value]} ->
        waiting = stream_spawn(value, spawned, waiting, monitor_pid, monitor_ref, timeout)
        stream_reduce({:cont, acc}, max - 1, spawned + 1, delivered, waiting, :done,
                      reducer, monitor_pid, monitor_ref, timeout, on_timeout)
      {_, []} ->
        stream_reduce({:cont, acc}, max, spawned, delivered, waiting, :done,
                      reducer, monitor_pid, monitor_ref, timeout, on_timeout)
    end
  end

  defp stream_deliver({:suspend, acc}, max, spawned, delivered, waiting, next,
                      reducer, monitor_pid, monitor_ref, timeout, on_timeout) do
    continuation = &stream_deliver(&1, max, spawned, delivered, waiting, next, reducer, monitor_pid, monitor_ref, timeout, on_timeout)
    {:suspended, acc, continuation}
  end
  defp stream_deliver({:halt, acc}, max, spawned, delivered, waiting, next,
                      reducer, monitor_pid, monitor_ref, timeout, on_timeout) do
    stream_reduce({:halt, acc}, max, spawned, delivered, waiting, next,
                  reducer, monitor_pid, monitor_ref, timeout, on_timeout)
  end
  defp stream_deliver({:cont, acc}, max, spawned, delivered, waiting, next,
                      reducer, monitor_pid, monitor_ref, timeout, on_timeout) do
    case waiting do
      %{^delivered => {nil, reply}} ->
        try do

        :ok = Process.cancel_timer(timer_ref, async: true, info: false)
        message_kind = if(timed_out?, do: :timed_out, else: :down)
        send(parent_pid, {message_kind, {monitor_ref, position}, reason})
        stream_monitor_loop(running_tasks, config)

      # One of the spawned processes timed out. We kill that process here
      # regardless of the value of :on_timeout. We then send a message to the
