refactor(stream): move duplicate code to function