This is called from inside the lock being held which means that replaying all historical values to a new Observer will block all existing Observers and new values from proceeding.