Extract common runStream function. observe and reduce were nearly identical, so it was easy to extract their guts into a common function to process a stream to completion