def notify(self, stream_key, stream_id, time_now_ms):
        self.current_token = self.current_token.copy_and_advance(
            stream_key, stream_id
        )
        self.last_notified_ms = time_now_ms
        noify_deferred = self.notify_deferred

        with PreserveLoggingContext():
            self.notify_deferred = ObservableDeferred(defer.Deferred())
            noify_deferred.callback(self.current_token)
Notify any listeners for this user of a new event from an
dataset/ETHPy150Open matrix-org/synapse/synapse/notifier.py/_NotifierUserStream.notify