stream: various fixes

* test: piping