为什么在所有初始订阅者断开连接后 RefCount 不起作用？

Question

考虑以下几点：

[Fact]
public void foo()
{
    var result = new Subject<bool>();
    var startCount = 0;
    var completionCount = 0;
    var obs = Observable
        .Defer(() =>
            {
                ++startCount;
                return result.FirstAsync();
            })
        .Do(_ => ++completionCount)
        .Publish()
        .RefCount();

    // pretend there are lots of subscribers at once
    var s1 = obs.Subscribe();
    var s2 = obs.Subscribe();
    var s3 = obs.Subscribe();

    // even so, we only expect to be started once
    Assert.Equal(1, startCount);
    Assert.Equal(0, completionCount);

    // and we won't complete until the result ticks through
    result.OnNext(true);
    Assert.Equal(1, startCount);
    Assert.Equal(1, completionCount);

    s1.Dispose();
    s2.Dispose();
    s3.Dispose();

    // now try exactly the same thing again
    s1 = obs.Subscribe();
    s2 = obs.Subscribe();
    s3 = obs.Subscribe();

    // startCount is 4 here instead of the expected 2!
    Assert.Equal(2, startCount);
    Assert.Equal(1, completionCount);

    result.OnNext(true);
    Assert.Equal(2, startCount);
    Assert.Equal(2, completionCount);

    s1.Dispose();
    s2.Dispose();
    s3.Dispose();
}

我对@9​​87654322@ + RefCount 的理解是，只要至少有一个订阅者，就会保持与源的连接。一旦最后一个订阅者断开连接，任何未来的订阅者都将重新启动与源的连接。

正如您在我的测试中看到的那样，第一次一切正常。但第二次，管道内的延迟 observable 会为每个新订阅者执行一次。

我可以通过调试器看到，对于第一组订阅者，obs._count（计算订阅者）在每次调用 Subscribe 时都会增加。但是对于第二组订阅者，它仍然是零。

为什么会发生这种情况，我可以做些什么来纠正我的管道？

Answer 1

@user631090 的答案很接近，但不正确，所以我想我会自己回答。

这是因为如果Publish 发布的流本身已经完成，它将立即完成新订阅者。你可以在图中看到here：

但如果图表包含订阅者在基础流完成之后会很好。

为了增加混乱，Defer 仍被称为新订阅者。但由于初始流完成，它的返回值被Publish 简单地忽略了。

我还没有想出一种方法来实现我的预期用例。我想也许使用Multicast 而不是Publish，根据需要创建一个新主题。但我还没有能够做到这一点。对于我认为常见的用例来说，这似乎相当痛苦。

Answer 2

这是因为底层的 observable result 已经完成了。所以每个新订阅者都只是得到 OnCompleted 回调。

如果 ObservableDefer 每次都创建一个新序列或一个未完成的序列，您会看到所需的行为。

例如

return result.FirstAsync().Concat(Observable.Never<bool>());

您需要删除Assert.Equal(1, completionCount);