Parallel Operator Rewrite

A rewrite of the `parallel` operator that is more efficient by not using `groupBy` and ready to support backpressure.