         {{counter, _, _}, counters} = Map.pop(counters, ref)
         send(parent_pid, {:down, {monitor_ref, counter}, reason})
         stream_monitor_loop(parent_pid, parent_ref, mfa, spawn, monitor_ref, counters)
+
       {:EXIT, _, _} ->
         stream_monitor_loop(parent_pid, parent_ref, mfa, spawn, monitor_ref, counters)
     end

    {:done, acc}
  end

  # No more tasks to spawned because max == 0 or next is done.
  defp stream_reduce({:cont, acc}, max, spawned, delivered, waiting, next,
                     reducer, monitor_pid, monitor_ref, timeout)
       when max == 0

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

