Added streaming support for Observable.