Tại sao RefCount không hoạt động sau khi tất cả người đăng ký ban đầu ngắt kết nối?

Question

xem xét như sau:

[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(); 
} 

sự hiểu biết của tôi về Publish + RefCount là một kết nối với nguồn được duy trì càng lâu càng có ít nhất một thuê bao. Khi người đăng ký cuối cùng ngắt kết nối, mọi người đăng ký trong tương lai sẽ bắt đầu lại kết nối với nguồn.

Như bạn thấy trong bài kiểm tra của tôi, mọi thứ hoạt động hoàn hảo trong lần đầu tiên. Nhưng lần thứ hai, sự quan sát được hoãn lại bên trong đường ống được thực hiện một lần cho mỗi người đăng ký mới.

Tôi có thể xem qua trình gỡ lỗi cho nhóm người đăng ký đầu tiên, obs._count (tính số người đăng ký) tăng cho mỗi cuộc gọi đến Subscribe. Nhưng đối với nhóm người đăng ký thứ hai, nó vẫn bằng không.

Tại sao điều này xảy ra và tôi có thể làm gì để khắc phục đường ống của mình?

Answer 1

Đó là vì kết quả có thể quan sát được kết quả đã hoàn tất. Vì vậy, mỗi người đăng ký mới chỉ nhận được cuộc gọi lại OnCompleted.

Nếu ObservableDefer đang tạo chuỗi mới mỗi lần hoặc một lần không hoàn thành, bạn sẽ thấy hành vi mong muốn.

ví dụ:

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

Bạn sẽ cần phải loại bỏ các Assert.Equal(1, completionCount);

Answer 2

Câu trả lời từ @ user631090 gần, nhưng không chính xác, vì vậy tôi nghĩ rằng tôi muốn trả lời bản thân mình.

Đó là vì Publish sẽ ngay lập tức hoàn tất người đăng ký mới nếu luồng được xuất bản tự hoàn thành. Bạn có thể phần nào thấy rằng trong sơ đồ here:

Nhưng nó sẽ được tốt đẹp nếu sơ đồ bao gồm một thuê bao sau dòng cơ bản hoàn tất.

Để thêm vào sự nhầm lẫn, Defer vẫn được gọi cho người đăng ký mới. Nhưng giá trị trả lại của nó chỉ đơn giản là bị bỏ qua bởi Publish vì luồng đầu tiên hoàn thành.

Tôi chưa thể tìm ra cách để triển khai trường hợp sử dụng dự định của mình. Tôi nghĩ có thể sử dụng Multicast thay vì Publish, tạo chủ đề mới nếu cần. Nhưng tôi chưa thể đạt được điều đó. Và nó có vẻ khá đau đớn cho những gì tôi nghĩ là một trường hợp sử dụng phổ biến.