Added ConsumerHandle to track subscriptions and allow user to cancel them.