updating how observable to stream works