Обходные пути для сбоев RX?

Обновление: позвольте мне попытаться ответить на мой собственный вопрос в контексте RX.

Во-первых, кажется, мое понимание того, что такое «глюк», было неправильным. С точки зрения чистого FRP то, что мне казалось сбоями в RX, на самом деле кажется правильным поведением в RX.

Поэтому я предполагаю, что в RX нам нужно четко указать «время», в которое мы ожидаем активировать значения, объединенные с датчиков.

В моем же примере приводом является консоль, а датчиком интервал n.

Итак, если я изменю свой код

ab.Subscribe(Console.WriteLine);

в

ab.Sample(n).Subscribe(Console.WriteLine);

то печатаются только "правильные" значения.

Это означает, что когда мы получаем наблюдаемую последовательность, которая объединяет значения от датчиков, мы должны знать все исходные датчики, объединять их все и пробовать значения с этим объединенным сигналом, прежде чем отправлять какие-либо значения на исполнительные механизмы...

Таким образом, альтернативным подходом было бы «поднять» IObservable в структуру «Sensed», которая запоминает и объединяет исходные датчики, например, так:

public struct Sensed<T>
{
    public IObservable<T> Values;
    public IObservable<Unit> Sensors;

    public Sensed(IObservable<T> values, IObservable<Unit> sensors)
    {
        Values = values;
        Sensors = sensors;
    }

    public IObservable<Unit> MergeSensors(IObservable<Unit> sensors)
    {
        return sensors == Sensors ? Sensors : Sensors.Merge(sensors);
    }

    public IObservable<T> MergeValues(IObservable<T> values)
    {
        return values == Values ? Values : Values.Merge(values);
    }
}

И тогда мы должны перевести весь метод RX в эту структуру «Sensed»:

public static class Sensed
{
    public static Sensed<T> Sensor<T>(this IObservable<T> source)
    {
        var hotSource = source.Publish().RefCount();
        return new Sensed<T>(hotSource, hotSource.Select(_ => Unit.Default));
    }

    public static Sensed<long> Interval(TimeSpan period)
    {
        return Observable.Interval(period).Sensor();
    }

    public static Sensed<TOut> Lift<TIn, TOut>(this Sensed<TIn> source, Func<IObservable<TIn>, IObservable<TOut>> lifter)
    {
        return new Sensed<TOut>(lifter(source.Values), source.Sensors);
    }

    public static Sensed<TOut> Select<TIn, TOut>(this Sensed<TIn> source, Func<TIn, TOut> func)
    {
        return source.Lift(values => values.Select(func));
    }

    public static Sensed<T> Where<T>(this Sensed<T> source, Func<T, bool> func)
    {
        return source.Lift(values => values.Where(func));
    }

    public static Sensed<T> Merge<T>(this Sensed<T> source1, Sensed<T> source2)
    {
        return new Sensed<T>(source1.MergeValues(source2.Values), source1.MergeSensors(source2.Sensors));
    }

    public static Sensed<TOut> CombineLatest<TIn1, TIn2, TOut>(this Sensed<TIn1> source1, Sensed<TIn2> source2, Func<TIn1, TIn2, TOut> func)
    {
        return new Sensed<TOut>(source1.Values.CombineLatest(source2.Values, func), source1.MergeSensors(source2.Sensors));
    }

    public static IDisposable Actuate<T>(this Sensed<T> source, Action<T> next) 
    {
        return source.Values.Sample(source.Sensors).Subscribe(next);
    }
}

Тогда мой пример становится:

var n = Sensed.Interval(TimeSpan.FromMilliseconds(100));
var a = n.Select(t => "a=" + t);
var b = n.Where(t => t % 10 == 0).Select(t => "b=" + t);
var ab = a.CombineLatest(b, Tuple.Create);
ab.Actuate(Console.WriteLine);

И снова на исполнительный механизм передаются только «желаемые» значения, но в этой конструкции исходные датчики запоминаются в структуре Sensed.

Я не уверен, что все это имеет «смысл» (каламбур), может быть, мне следует просто отпустить свое желание получить чистый FRP и жить с этим. Ведь время относительно ;-)

Питер