continues refactor and adds ability to stop stream