为什么在这个 Suspendable 实现（“Rx Pausable”）中没有调用 OnCompleted？答案

【问题标题】：Why is not OnCompleted not called in this Suspendable implementation ("Rx Pausable")?为什么在这个 Suspendable 实现（“Rx Pausable”）中没有调用 OnCompleted？
【发布时间】：2014-10-24 19:43:01
【问题描述】：

我阅读了以下从 Ollie Riches 的 博客帖子 Trying to be more functional with Rx 中截取的内容，并开始和作者一样想知道：为什么 OnCompleted 没有通过？有人可以告诉这里发生了什么吗？也许是简单得令人尴尬的事情？

为了方便起见，这里对代码进行了一些修改和复制（如果不能在这里撕掉他的代码，我向 Ollie 道歉）：

public static class RxExtensions
{
    public static IObservable<T> Suspendable<T>(this IObservable<T> stream, IObservable<bool> suspend, bool initialState = false)
    {
        return Observable.Create<T>(o =>
        {
            var disposable = suspend.StartWith(initialState)
                    .DistinctUntilChanged()
                    .Select(s => s ? Observable.Empty<T>() : stream)
                    .Switch()
                    .Subscribe(o);

            return disposable;
        });
    }
}

var testScheduler = new TestScheduler();
var generatorCount = 10;

//If the limit will be hardcoded to something less than generatorCount, an exception will be
//thrown and the exception object will be set. Why it doesn't happen to completed in the following?
var generator = Observable.Generate(1,
    x => x <= generatorCount,
    x => x + 1,
    x => { if(x != 11) { Console.WriteLine(x); return x; } else { throw new ArgumentException(); } },
    x => TimeSpan.FromSeconds(1),
    testScheduler);


Exception exception = null;
var completed = false;
generator.Suspendable(new Subject<bool>()).Subscribe(_ => { }, e => exception = e, () => completed = true);   
testScheduler.AdvanceBy(TimeSpan.FromMilliseconds(1001000).Ticks);

Console.WriteLine(exception);
Console.WriteLine(completed);

为了记录，我正在考虑尝试生成一个可以暂停和停止的流，区别在于暂停流累积事件，暂停只是跳过它们。它开始看起来比我预期的要复杂一些，特别是如果有人想对暂停的位设置限制或“保存策略”。哦，好吧……

有趣的是，我刚刚注意到RxJS implementation of Pausable。

【问题讨论】：

标签： c# system.reactive rxjs

【解决方案1】：

您的观察者同时订阅了suspend 流和source 流。在两个流都完成之前，该组合流不会完成。基本上您的source 流已完成，但Suspendable 正在等待查看是否有更多暂停/取消暂停信号会通过。如果他们这样做，它将重新订阅源流。

在源流完成时让可暂停流完成是可能的，但可能会破坏您的方法的目的。基本上必须保持订阅源流并在源完成时结束暂停的流。你可以这样做：

var shared = stream.Publish();
var pausable = suspend
    .StartWith(initialState)
    .TakeUntil(shared.LastOrDefaultAsync())
    .DistinctUntilChanged()
    .Select(p => p ? shared : Observable.Empty<T>())
    .Switch();
var disposable = new CompositeDisposable(pausable.Subscribe(o), shared.Connect());
return disposable;

【讨论】：

嘿，所以，这是一个简单的问题。这是对正在发生的事情的一个很好的解释。我刚刚注意到一个关于这个Pause and Resume Subscription on cold IObservable 的问题。这种方法感觉有点像对流施加背压的方法，不是那么简单，只要有更花哨的东西。

【解决方案2】：

Completed 未发送，因为您的订阅在 Observable.Empty() 上，而不是您的 _generator 的后代

所以我给你一个更好的答案，使用 CombineLatest

public static IObservable<T> Suspendable<T>(
    this IObservable<T> source,
    IObservable<bool> pauser,
    bool initialState = false)
{
    return 
        source.CombineLatest(pauser.StartWith(initialState), 
                             (value, paused) => new {value, paused})
              .Where(_=>!_.paused)
              .Select(_=>_.value);
}

【讨论】：