 
   defp flush_events(ref) do
     receive do
-      {:notify, _event} ->
+      {_from, {_pid, ^ref}, {:notify, _event}} ->
         flush_events(ref)
     after
       0 -> :ok

        {:halt, acc}

      # Got an async event.
      {:notify, event} ->
        {[{:async, pid, ref, event}], acc}

      # Got a sync event.

    {handlers, streams}
  end

  defp server_stream_notify(mode, event, handler(pid: pid, ref: ref)) do
    send pid, {self(), {self(), ref}, {mode_to_tag(mode), event}}
  end

  defp mode_to_tag(:ack),   do: :ack_notify
  defp mode_to_tag(:sync),  do: :sync_notify
  defp mode_to_tag(:async), do: :notify

  defp server_notify(event, fun, [handler|t], name, handlers, acc, hib) do
    case server_update(handler, fun, event, name, handlers) do
