asynchshell: refactored pipe_consume and pipe_finish