ストリームのカスタマイズ可能なフィルタリング

まず、入力は次のようになります。

var input = scheduler.CreateColdObservable<char>(
    ReactiveTest.OnNext(0100.Ms(), '1'),
    ReactiveTest.OnNext(0200.Ms(), 'A'),
    ReactiveTest.OnNext(0300.Ms(), '2'),
    ReactiveTest.OnNext(0400.Ms(), 'B'), //You forgot this line
    ReactiveTest.OnNext(0500.Ms(), 'A'),
    ReactiveTest.OnNext(0600.Ms(), 'B'),
    ReactiveTest.OnNext(0700.Ms(), '5'),
    ReactiveTest.OnNext(0800.Ms(), 'A'),
    ReactiveTest.OnNext(0900.Ms(), 'C'));

また、@ LeeCampbellが述べたように、Ms拡張メソッドはタイプlongではなくタイプを返す必要がありTimeSpanます。

これが私が思いついた解決策です：

 var activeFilters = filters
    .Select(s => s.Split(' '))
    .Select(s => Tuple.Create(s[0][0], TimeSpan.FromMilliseconds(int.Parse(s[1]))))
    .Select(t => Observable.Timer(t.Item2, scheduler).Select(_ => t.Item1).StartWith(t.Item1))
    .MergeCombineLatest(true)
    .StartWith(new List<char>());

var output = activeFilters.Publish(_activeFilters => 
        input.Join(_activeFilters,
            _ => Observable.Return(1),
            t => _activeFilters,
            (c, filterList) => Tuple.Create(c, filterList)
        )
    )
    .Where(t => !t.Item2.Contains(t.Item1))
    .Select(t => t.Item1);

var observer = scheduler.CreateObserver<char>();
output.Subscribe(observer);
scheduler.Start();

ReactiveAssert.AreElementsEqual(
    expected.Messages,
    observer.Messages);

activeFilters現在フィルタリング中の文字のリストを発行するオブザーバブルです。アクティブにフィルタリングされたリストが変更されるとactiveFilters、新しいリストが発行されます。同じ文字に複数のフィルターを含めることができるため、リストは必ずしも一意ではないことに注意してください。

どのフィルターがアクティブであるかをいつでも把握できたら、そのリストを入力に結合できます。

コードには、System.Collections.ImmutableNugetパッケージを使用する次の拡張メソッドが必要です。

public static IObservable<IList<T>> MergeCombineLatest<T>(this IObservable<IObservable<T>> outer, bool removeCompleted)
{
    return outer
        .SelectMany((inner, i) => inner
            .Materialize()
            .SelectMany(nt => nt.Kind == NotificationKind.OnNext
                ? Observable.Return(Tuple.Create(i, nt.Value, true))
                : nt.Kind == NotificationKind.OnCompleted
                    ? removeCompleted
                        ? Observable.Return(Tuple.Create(i, default(T), false))
                        : Observable.Empty<Tuple<int, T, bool>>()
                    : Observable.Throw<Tuple<int, T, bool>>(nt.Exception)
            )
        )
        .Scan(ImmutableDictionary<int, T>.Empty, (dict, t) => t.Item3 ? dict.SetItem(t.Item1, t.Item2) : dict.Remove(t.Item1))
        .Select(dict => dict.Values.ToList());
}

MergeCombineLatestオブザーバブルのオブザーバブルを取得し、各子オブザーバブルから最新の値のリストを発行します。場合はremoveCompletedtrueで、そのとき子供に観察が完了すると、リストは1収縮します。removeCompletedfalseの場合、最後の値はリストと後続のリストに永久に残ります。

これを行うためのより友好的な方法があれば、私は多くの義務を負います。