         :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(parent_pid, parent_ref, mfa, spawn, monitor_ref, running_tasks, timeout)
+        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

    # about our reference to it.
    send(monitor_pid, {parent, monitor_ref})

    stream_reduce(acc, max_concurrency, _spawned = 0, _delivered = 0, _waiting = %{}, next,
                  reducer, monitor_pid, monitor_ref, timeout, on_timeout)
  end

  defp stream_reduce({:halt, acc}, _max, _spawned, _delivered, _waiting, next,
                     _reducer, monitor_pid, monitor_ref, timeout, _on_timeout) do
    stream_close(monitor_pid, monitor_ref, timeout)
    is_function(next) && next.({:halt, []})
    {:halted, acc}
  end

  defp stream_reduce({:suspend, acc}, max, spawned, delivered, waiting, next,
                     reducer, monitor_pid, monitor_ref, timeout, on_timeout) do
    continuation = &stream_reduce(&1, max, spawned, delivered, waiting, next, reducer, monitor_pid, monitor_ref, timeout, on_timeout)
    {:suspended, acc, continuation}
  end

  # All spawned, all delivered, next is :done.
  defp stream_reduce({:cont, acc}, _max, spawned, delivered, _waiting, next,
                     _reducer, monitor_pid, monitor_ref, timeout, _on_timeout)
       when spawned == delivered and next == :done do
    stream_close(monitor_pid, monitor_ref, timeout)
    {:done, acc}

            _other ->
              running_tasks
          end
        stream_monitor_loop(running_tasks, config)

      {:EXIT, _, _} ->
        stream_monitor_loop(running_tasks, config)
    end
  end

