Add support to use by() like a regular stream