refactor as pull-stream