'Notify any listeners for this user of a new event from an
event source.
Args:
stream_key(str): The stream the event came from.
stream_id(str): The new id for the stream the event came from.
time_now_ms(int): The current time in milliseconds.'

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_token = self.current_token
    self.last_notified_ms = time_now_ms
    noify_deferred = self.notify_deferred
    users_woken_by_stream_counter.inc(stream_key)
    with PreserveLoggingContext():
        self.notify_deferred = ObservableDeferred(defer.Deferred())
        noify_deferred.callback(self.current_token)
