Fix consumer hang on connection errors

The consumer tended to get hung when connection errors occurred. This commit re-implements the consumer based on the asynchronous consumer example.