refactor the pipe stream