Rx 缓存 - 重播运算符清除答案

【问题标题】：Rx Cache - Replay operator ClearRx 缓存 - 重播运算符清除
【发布时间】：2016-05-20 03:44:51
【问题描述】：

我正在使用此处的以下代码 - 在清除“重放缓存”时对我来说似乎是个问题

https://gist.github.com/leeoades/4115023

如果我像这样更改以下调用和代码，我会发现 Replay 中存在错误，即它永远不会被清除。有人可以帮忙解决这个问题吗？

private Cache<string> GetCalculator()
    {
        var calculation = Observable.Create<string>(o =>
        {
            _calculationStartedCount++;

            return Observable.Timer(_calculationDuration, _testScheduler)
                             .Select(_ => "Hello World!" + _calculationStartedCount) // suffixed the string with count to test the behaviour of Replay clearing
                             .Subscribe(o);
        });

        return new Cache<string>(calculation);
    }

[Test]
    public void After_Calling_GetResult_Calling_ClearResult_and_GetResult_should_perform_calculation_again()
    {
        // ARRANGE
        var calculator = GetCalculator();

        calculator.GetValue().Subscribe();
        _testScheduler.Start();

        // ACT
        calculator.Clear();

        string result = null;
        calculator.GetValue().Subscribe(r => result = r);
        _testScheduler.Start();

        // ASSERT
        Assert.That(_calculationStartedCount, Is.EqualTo(2));
        Assert.That(result, Is.EqualTo("Hello World!2")); // always returns Hello World!1 and not Hello World!2
        Assert.IsNotNull(result);
    }

【问题讨论】：

标签： c# .net system.reactive

【解决方案1】：

这个问题很微妙。源序列Timer 在它发出一个事件后完成，该事件又在Replay 创建的内部ReplaySubject 上调用OnCompleted。当Subject 完成时，即使出现新的Observable，它也不再接受任何新值。

当您重新订阅底层Observable 时，它会再次执行，但无法重新启动Subject，因此您的新Observer 只能接收ReplaySubject 完成之前的最新值。

最简单的解决方案可能就是永远不要让源流完成（未经测试）：

    public Cache(IObservable<T> source)
    {
        //Not sure why you are wrapping this in an Observable.create
        _source = source.Concat(Observable.Never())
                            .Replay(1, Scheduler.Immediate);
    }

【讨论】：