Use stream-consume to ensure stream completes

Move stream-consuming logic to new stream-consume module.