Perché RefCount non funziona dopo che tutti gli abbonati iniziali si sono scollegati?

Question

Si consideri il seguente:

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

mia comprensione di Publish + RefCount è che una connessione alla sorgente viene mantenuta finché c'è almeno un utente. Una volta che l'ultimo utente si disconnette, qualsiasi utente futuro riavvierà la connessione alla fonte.

Come potete vedere nel mio test, tutto funziona alla perfezione. Ma la seconda volta, il rinviabile osservabile all'interno della pipeline viene eseguito una volta per ogni nuovo abbonato.

Vedo tramite il debugger che per il primo gruppo di abbonati, obs._count (che conta gli abbonati) aumenta per ogni chiamata a Subscribe. Ma per il secondo gruppo di abbonati, rimane zero.

Perché sta succedendo questo e cosa posso fare per rettificare la mia pipeline?

Answer 1

È perché il risultato osservabile sottostante è già stato completato. Quindi ogni nuovo iscritto ottiene solo il callback OnCompleted.

Se ObservableDefer stava creando una nuova sequenza ogni volta o uno che non ha completato si dovrebbe vedere il comportamento desiderato.

ad es.

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

Sarà necessario rimuovere il Assert.Equal(1, completionCount);

Answer 2

La risposta da @ user631090 è vicino, ma non corretto, così ho pensato di rispondere io.

È perché Publish completerà immediatamente nuovi abbonati se il flusso ha pubblicato è in sé compiuto. È possibile tipo di vedere che nel diagramma here:

ma sarebbe stato bello se il diagramma incluso un abbonato dopo completa il flusso sottostante.

Per aumentare la confusione, Defer è ancora chiamato per i nuovi abbonati. Ma il suo valore di ritorno viene semplicemente ignorato da Publish a causa dello streaming iniziale che termina.

io sono ancora in grado di venire con un modo per implementare il mio caso destinazione d'uso. Ho pensato forse di usare Multicast anziché Publish, creando un nuovo soggetto, se necessario. Ma non sono ancora riuscito a farlo. E sembra piuttosto doloroso per quello che penserei sia un caso di uso comune.