Convert filter from an operator to a processor

[reubano]

There's no reason why filter needs access to the entire stream. It should be able to operate on individual items.