Dlaczego RefCount nie działa po rozłączeniu wszystkich początkowych subskrybentów?

Question

Rozważmy następujący:

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

Moje rozumienie Publish + RefCount jest to, że podłączenie do źródła utrzymuje się tak długo, jak istnieje co najmniej jeden abonent. Po rozłączeniu ostatniego subskrybenta każdy przyszły subskrybent ponownie zainicjuje połączenie ze źródłem.

Jak widać w moim teście, wszystko działa idealnie za pierwszym razem. Ale drugi raz, odroczony obserwowalny wewnątrz potoku jest wykonywany raz dla każdego nowego abonenta.

Widzę przez debugger, że dla pierwszej grupy subskrybentów, obs._count (który liczy abonentów) wzrasta dla każdego połączenia do Subscribe. Ale dla drugiej grupy abonentów pozostaje ona zerowa.

Dlaczego tak się dzieje i co mogę zrobić, aby naprawić mój rurociąg?

Answer 1

Jest tak, ponieważ podstawowy obserwowalny wynik zakończył się. Tak więc każdy nowy subskrybent otrzymuje właśnie wywołanie zwrotne OnCompleted.

Jeśli ObservableDefer tworzył nową sekwencję za każdym razem lub taką, która nie została zakończona, zobaczysz pożądane zachowanie.

np.

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

Trzeba będzie usunąć Assert.Equal(1, completionCount);

Answer 2

odpowiedzi z @ user631090 jest blisko, ale błędne, więc myślałem, że sobie odpowiedzieć.

To dlatego, że Publish natychmiast ukończy nowych subskrybentów, jeśli strumień, który opublikował sam zakończył. niby można zobaczyć, że na wykresie here:

Ale byłoby miło, gdyby schemat zawarte abonenta po podstawowa strumień kończy.

Aby dodać do zamieszania, Defer jest nadal wezwany dla nowych subskrybentów. Ale jego wartość zwracana jest po prostu ignorowana przez Publish z powodu początkowego zakończenia strumienia.

Nie jestem jeszcze w stanie wymyślić sposobu na wdrożenie mojego przypadku użycia. Pomyślałem, że może używając Multicast zamiast Publish, w razie potrzeby tworząc nowy temat. Ale nie udało mi się tego jeszcze osiągnąć. I wydaje się raczej bolesne dla tego, co uważam za powszechny przypadek użycia.