       {:gen_event_EXIT, {__MODULE__, ^ref}, _reason} = event ->
         Process.demonitor(ref, [:flush])
         send(self(), event)
-        {:halt, acc}
+        {:halt, {:removed, acc}}
 
       # The manager died. Stop iteration, resolve the event later.
       {:DOWN, ^ref, _, _, _} = event ->
         send(self(), event)
-        {:halt, acc}
+        {:halt, {:removed, acc}}
 
       # Duration timeout.
       {^ref, :timedout} ->

      {:error, reason} ->
        exit({reason, {__MODULE__, :stop, [stream, acc]}})
    end
  end

  defp remove_timer(nil), do: :ok

    end
  end

  # If we reach this branch, we know the handler was already
  # removed, so we don't trigger a request for doing so.
  defp stop(%{mode: mode} = stream, {:removed, {pid, ref, timer} = acc}) do
    remove_timer(timer)

    case wait_for_handler_removal(pid, ref) do
      :ok when mode == :async ->
