converted source() and each() to filter/pipe form
fromStream takes an optional output channel, optionally left open