refactor(stream) stream -> pipe