Add observe() operation to passively consume a stream
Add overflow() operation to overflow to a specified completable queue e.g. for fault tolerance support.