Csharp Rx.Net窗口内部可观测数据提前完成

发布于02月16日

为什么

Observable.Interval(TimeSpan.FromSeconds(1)).Window(2, 1).Concat()

放弃0,1,2，……？

我的预期是，.Window(2, 1)将产生一系列数字对(0, 1、1, 2、2, 3、...)，因此.Concat()将一次产生两个数字，而不是一个.

我注意到

Observable.Interval(TimeSpan.FromSeconds(1)).Window(2, 1).Merge()

does按预期运行，然而在我的用例中，我有.Select()在.Window()和.Concat()之间执行工作，并且重要的是我保持了原始序列的顺序(在我的实际代码中实际上不是Observable.Interval)，因此希望使用.Concat()来代替.

我的误解是什么导致了这种差异，我如何才能达到我想要的行为？

推荐答案

因为.Concat()只订阅当前完成后的下一个内部可观察对象，并且在订阅时，重叠的"下一个"可观察对象已经发出了它的第一个值，没有人消费.

使用.Merge()或或.SelectMany(x => x)而不是合并或使用.Buffer 或者创建如下所示的缓冲操作符:

public static class RxExtensions
{
    public static IObservable<T> CollectUntilSubscribed<T, TUntil>(this IObservable<T> source, IObservable<TUntil> stopMe)
        {
            var subscribersCount = 0;

            var ready = new Subject<Unit>();

            var delayed = source
                .Delay(x => subscribersCount > 0 ? Observable.Return(Unit.Default):ready)
                .Publish()
                ;

            var connection = delayed.Connect();

            stopMe.Take(1).Subscribe(_ => connection.Dispose());

            return Observable.Create<T>(o =>
            {
                var sub = delayed.Subscribe(o);

                if (Interlocked.Increment(ref subscribersCount) == 1) ready.OnNext(Unit.Default);

                return Disposable.Create(() =>
                {
                    Interlocked.Decrement(ref subscribersCount);
                    sub.Dispose();
                });
            });
        }
}

internal class Program
{
    static void Main(string[] args)
    {
        Observable
            .Interval(TimeSpan.FromSeconds(1))
            .Window(2,1)
            .Select(x => x.CollectUntilSubscribed(Observable.Never<Unit>()))
            .Concat()
            .Subscribe(Console.WriteLine);

            Console.ReadLine();
    }
}