Add a high level API for consuming messages

Provide a high-level API for reading messages delivered to a consumer.