IObservable.FirstOrDefaultObservable() - a non-blocking FirstOrDefault for IObservable

gistfile1.cs

    public static class ObservableExtensions
    {
        public static IObservable<T> FirstOrDefaultObservable<T>(this IObservable<T> source)
        {
            return Observable
                .Create<T>(
                    observer =>
                    {
                        bool hasFired = false;
                        IDisposable disposable = null;

                        disposable = source.Subscribe(
                            next =>
                            {
                                if (hasFired == false)
                                { 
                                    hasFired = true;
                                    observer.OnNext(next); 
                                    observer.OnCompleted();
                                    disposable.Dispose();
                                }
                            },
                            error => observer.OnError(error),
                            () =>
                            {
                                if (false == hasFired)
                                {
                                    observer.OnNext(default(T));
                                    observer.OnCompleted();
                                }
                            });

                        return () => disposable.Dispose();
                    });
        }
    }