Moved _run_changed inside run_changed like other amqp consumers.