   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})
+  defp stream_reduce({:cont, acc}, _max, spawned, delivered, _waiting, next, config)
        when spawned == delivered and next == :done do
+    %{
+      monitor_pid: monitor_pid,
+      monitor_ref: monitor_ref,
+      timeout: timeout
+    } = config
+
     stream_close(monitor_pid, monitor_ref, timeout)
     {:done, acc}
   end


  def reply(caller, monitor, info, mfa) do
    initial_call(mfa)
    case monitor do
      :monitor ->
        mref = Process.monitor(caller)
        reply(caller, mref, @ref_timeout, info, mfa)
      :nomonitor ->
        reply(caller, nil, :infinity, info, mfa)
    end

  end

  defp stream_deliver({:cont, acc}, max, spawned, delivered, waiting, next, config) do
    %{
      reducer: reducer,
      monitor_pid: monitor_pid,
      monitor_ref: monitor_ref,
      timeout: timeout
    } = config

    case waiting do
      %{^delivered => {:done, reply}} ->
