hoge.cs

namespace NcmbAsObservables
{
    /// <summary>
    /// NCMBObjectの拡張
    /// </summary>
    public static class NcmbObjectExtensions
    {
        /// <summary>
        /// 非同期処理でオブジェクトの取得を行います。
        /// </summary>
        /// <returns>取得したオブジェクト</returns>
        public static IObservable<NCMBObject> FetchAsyncAsStream(this NCMBObject origin)
        {
            return Observable.Create<NCMBObject>(observer =>
            {
                var disposable = new BooleanDisposable();
                origin.FetchAsync(error =>
                {
                    if (disposable.IsDisposed) return;
                    if (error == null)
                    {
                        observer.OnNext(origin);
                        observer.OnCompleted();
                    }
                    else
                    {
                        observer.OnError(error);
                    }
                });
                return disposable;
            });
        }
    }
}