ConsumerStream with async-stream instead of custom polling

[fraidev]

Based on #4405 (comment).
We should change ConsumerStream implementations to use something better than custom polling.

Maybe https://crates.io/crates/async-stream