ThrowIfFalse 条件式がFalseを返したら例外を発行するIObservable<T> 拡張メソッド

gistfile1.txt

using System;
using System.Collections.Generic;
using System.Linq;
using System.Reactive.Linq;

namespace ConsoleApplication5
{

    class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine();
            Console.WriteLine("[条件式がFalseを返す場合に例外を後続へ流す]");
            Observable.Range(1, 5)
                .ThrowIfFalse(x => x < 4,
                    new Exception("x>=3"))
                .Subscribe(
                    x => Console.WriteLine($"OnNext({x})"),
                    e => Console.WriteLine($"OnError({e.Message})"));

            Console.WriteLine();
            Console.WriteLine("[例外をSelectorで指定]");
            Observable.Range(1, 5)
                .ThrowIfFalse(x => x < 4,
                    x => new Exception($"{x} [NG!]"))
                .Subscribe(
                    x => Console.WriteLine($"OnNext({x})"),
                    e => Console.WriteLine($"OnError({e.Message})"));

            Console.WriteLine();
            Console.WriteLine("[加工して後続へ流す]");
            Observable.Range(1, 5)
                .ThrowIfFalse(x => x < 4,
                    new Exception("x>=3"),
                    x => x * x)
                .Subscribe(
                    x => Console.WriteLine($"OnNext({x})"),
                    e => Console.WriteLine($"OnError({e.Message})"));

            Console.WriteLine();
            Console.WriteLine("[False時の値を流してから例外を流す]");
            Observable.Range(1, 5)
                .ThrowIfFalse(x => x < 4,
                    new Exception("x>=3"),
                    x => x,
                    x => x)
                .Subscribe(
                    x => Console.WriteLine($"OnNext({x})"),
                    e => Console.WriteLine($"OnError({e.Message})"));
            Console.Read();
        }
    }
}

//[条件式がFalseを返す場合に例外を後続へ流す]
//OnNext(1)
//OnNext(2)
//OnNext(3)
//OnError(x>=3)

//[例外をSelectorで指定]
//OnNext(1)
//OnNext(2)
//OnNext(3)
//OnError(4 [NG!])

//[加工して後続へ流す]
//OnNext(1)
//OnNext(4)
//OnNext(9)
//OnError(x>=3)

[False時の値を流してから例外を流す]
OnNext(1)
OnNext(2)
OnNext(3)
OnNext(4)
OnError(x>=3)

ObservableExt.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Reactive.Linq;

namespace ConsoleApplication5
{
static class ObservalbeExt
    {
        /// <summary>
        /// <paramref name="predicate"/> が falseを返す場合に、指定した例外を後続に流します。 
        /// true を返す場合は現在の値をそのまま後続へ流します。
        /// </summary>
        public static IObservable<T> ThrowIfFalse<T>(this IObservable<T> source, Func<T, bool> predicate, Exception exception)
        {
            ThrowIfNull(source, nameof(source));
            ThrowIfNull(predicate, nameof(predicate));
            ThrowIfNull(exception, nameof(exception));

            return _ThrowIfFalse(source, predicate, exception);
        }

        /// <summary>
        /// <paramref name="predicate"/> が falseを返す場合に、例外を後続に流します。例外は、例外オブジェクトを返す述語関数<paramref name="exceptionSelector"/>によって指定します。
        /// true を返す場合は現在の値をそのまま後続へ流します。
        /// </summary>
        public static IObservable<T> ThrowIfFalse<T, TException>(this IObservable<T> source, Func<T, bool> predicate,
            Func<T, TException> exceptionSelector) where TException : Exception
        {
            ThrowIfNull(source, nameof(source));
            ThrowIfNull(predicate, nameof(predicate));
            ThrowIfNull(exceptionSelector, nameof(exceptionSelector));

            return _ThrowIfFalse(source, predicate, exceptionSelector);
        }

        /// <summary>
        /// <paramref name="predicate"/> が falseを返す場合に、例外を後続に流します。例外は、例外オブジェクトを返す関数<paramref name="exceptionSelector"/>によって指定します。
        /// true を返す場合は <paramref name="resultSelector"/>によって変換した値を後続へ流します。
        /// </summary>
        public static IObservable<TResult> ThrowIfFalse<TSource, TResult, TException>(this IObservable<TSource> source, Func<TSource, bool> predicate,
            Func<TSource, TException> exceptionSelector, Func<TSource, TResult> resultSelector) where TException : Exception
        {
            ThrowIfNull(source, nameof(source));
            ThrowIfNull(predicate, nameof(predicate));
            ThrowIfNull(exceptionSelector, nameof(exceptionSelector));
            ThrowIfNull(resultSelector, nameof(resultSelector));

            return _ThrowIfFalse(source, predicate, exceptionSelector, resultSelector);
        }

        /// <summary>
        /// <paramref name="predicate"/> が falseを返す場合に、指定した例外を後続に流します。
        /// true を返す場合は <paramref name="resultSelector"/>によって変換した値を後続へ流します。
        /// </summary>
        public static IObservable<TResult> ThrowIfFalse<TSource, TResult>(this IObservable<TSource> source, Func<TSource, bool> predicate,
            Exception exception, Func<TSource, TResult> resultSelector)
        {
            ThrowIfNull(source, nameof(source));
            ThrowIfNull(predicate, nameof(predicate));
            ThrowIfNull(exception, nameof(exception));
            ThrowIfNull(resultSelector, nameof(resultSelector));

            return _ThrowIfFalse(source, predicate, exception, resultSelector);
        }

        /// <summary>
        /// <paramref name="predicate"/> が falseを返す場合、現在の値を変換関数 <paramref name="resultSelectorWhenFalse"/> によって変換した値を流した後、例外を後続に流します。
        /// true を返す場合は 変換関数 <paramref name="resultSelectorWhenTrue"/> によって変換した値を後続へ流します。
        /// </summary>
        public static IObservable<TResult> ThrowIfFalse<TSource, TResult>(this IObservable<TSource> source, Func<TSource, bool> predicate,
            Exception exception, Func<TSource, TResult> resultSelectorWhenTrue, Func<TSource, TResult> resultSelectorWhenFalse)
        {
            ThrowIfNull(source, nameof(source));
            ThrowIfNull(predicate, nameof(predicate));
            ThrowIfNull(exception, nameof(exception));
            ThrowIfNull(resultSelectorWhenTrue, nameof(resultSelectorWhenTrue));
            ThrowIfNull(resultSelectorWhenFalse, nameof(resultSelectorWhenFalse));

            return _ThrowIfFalse(source, predicate, exception, resultSelectorWhenTrue, resultSelectorWhenFalse);
        }

        /// <summary>
        /// <paramref name="predicate"/> が falseを返す場合、現在の値を変換関数<paramref name="resultSelectorWhenFalse"/>  によって変換した値を流した後、指定した例外を後続に流します。
        /// 例外は、例外オブジェクトを返す関数<paramref name="exceptionSelector"/> によって指定します。
        /// true を返す場合は 変換関数 <paramref name="resultSelectorWhenTrue"/> によって変換した値を後続へ流します。
        /// </summary>
        public static IObservable<TResult> ThrowIfFalse<TSource, TResult, TException>(this IObservable<TSource> source, Func<TSource, bool> predicate,
            Func<TSource, TException> exceptionSelector, Func<TSource, TResult> resultSelectorWhenTrue, Func<TSource, TResult> resultSelectorWhenFalse) where TException : Exception
        {
            ThrowIfNull(source, nameof(source));
            ThrowIfNull(predicate, nameof(predicate));
            ThrowIfNull(exceptionSelector, nameof(exceptionSelector));
            ThrowIfNull(resultSelectorWhenTrue, nameof(resultSelectorWhenTrue));
            ThrowIfNull(resultSelectorWhenFalse, nameof(resultSelectorWhenFalse));

            return _ThrowIfFalse(source, predicate, exceptionSelector, resultSelectorWhenTrue, resultSelectorWhenFalse);
        }


        private static IObservable<T> _ThrowIfFalse<T>(this IObservable<T> source, Func<T, bool> predicate, Exception exception)
        {
            return source.SelectMany(x => predicate(x) ? Observable.Return(x) : Observable.Throw<T>(exception));
        }


        private static IObservable<T> _ThrowIfFalse<T, TException>(this IObservable<T> source,
            Func<T, bool> predicate, Func<T, TException> exceptionSelector) where TException : Exception
        {
            return source.SelectMany(x => predicate(x) ?
                    Observable.Return(x) :
                    Observable.Throw<T>(exceptionSelector(x)));
        }

        private static IObservable<TResult> _ThrowIfFalse<TSource, TResult>(this IObservable<TSource> source, Func<TSource, bool> predicate,
            Exception exception, Func<TSource, TResult> resultSelector)
        {

            return source.SelectMany(x => predicate(x) ?
                    Observable.Return(resultSelector(x)) :
                    Observable.Throw<TResult>(exception));

        }

        private static IObservable<TResult> _ThrowIfFalse<TSource, TResult, TException>(this IObservable<TSource> source, Func<TSource, bool> predicate,
            Func<TSource, TException> exceptionSelector, Func<TSource, TResult> resultSelector) where TException : Exception
        {
            return source.SelectMany(x => predicate(x) ?
                    Observable.Return(resultSelector(x)) :
                    Observable.Throw<TResult>(exceptionSelector(x)));
        }


        private static IObservable<TResult> _ThrowIfFalse<TSource, TResult>(this IObservable<TSource> source, Func<TSource, bool> predicate,
            Exception exception, Func<TSource, TResult> resultSelectorWhenTrue, Func<TSource, TResult> resultSelectorWhenFalse)
        {
            return source.SelectMany(x => predicate(x) ?
                    Observable.Return(resultSelectorWhenTrue(x)) :
                    Observable.Return(resultSelectorWhenFalse(x)).Concat(Observable.Throw<TResult>(exception)));
        }

        private static IObservable<TResult> _ThrowIfFalse<TSource, TResult, TException>(this IObservable<TSource> source, Func<TSource, bool> predicate,
            Func<TSource, TException> exceptionSelector, Func<TSource, TResult> resultSelectorWhenTrue, Func<TSource, TResult> resultSelectorWhenFalse) where TException : Exception
        {
            return source.SelectMany(x => predicate(x) ?
                    Observable.Return(resultSelectorWhenTrue(x)) :
                    Observable.Return(resultSelectorWhenFalse(x)).Concat(Observable.Throw<TResult>(exceptionSelector(x))));
        }


        private static void ThrowIfNull(object obj, string name)
        {
            if (obj == null) { throw new ArgumentNullException(name); }
        }
    }
}