minor refactoring, moved custom transforms at the beginning of stream pipes