Changed to use IObservable streams