             :erlang.raise(kind, reason, stacktrace)
         else
           pair ->
-            stream_deliver(pair, max, spawned, delivered + 1, Map.delete(waiting, delivered), next,
-                           reducer, monitor_pid, monitor_ref, timeout, on_timeout)
+            stream_deliver(pair, max, spawned, delivered + 1, Map.delete(waiting, delivered), next, config)
         end
       %{} ->
-        stream_reduce({:cont, acc}, max, spawned, delivered, waiting, next,
-                      reducer, monitor_pid, monitor_ref, timeout, on_timeout)
+        stream_reduce({:cont, acc}, max, spawned, delivered, waiting, next, config)
     end
   end
 

        :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)

      # 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})

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

  defp stream_reduce({:halt, acc}, _max, _spawned, _delivered, _waiting, next, config) do
    %{monitor_pid: monitor_pid, monitor_ref: monitor_ref, timeout: timeout} = config
    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, config) do
    continuation = &stream_reduce(&1, max, spawned, delivered, waiting, next, config)
    {:suspended, acc, continuation}
  end

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