add blocking_observable and last, count operators