add Consumer.cancel() and refactor little bit
return next_message() iterator directly