Per i principianti in genere non è necessario implementare gli oggetti IObservable e IObserver. La tua domanda è di suggerimenti sul perché. È molto difficile ottenere il comportamento sottostante corretto.
Ora ogni chiamata a IObservable.Subscribe restituisce un IDisposable. Viene utilizzato se il chiamante di Subscribe desidera annullare l'iscrizione a un osservabile prima che venga chiamato uno OnCompleted o OnError. Tuttavia, se viene chiamato uno OnCompleted o OnError, lo IDisposable viene automaticamente eliminato. Così efficacemente Rx si ripulisce automaticamente da solo quando una collezione osservabile completa.
Ogni singolo osservatore non ha bisogno di gestire la propria durata di abbonamento. L'osservatore deve solo rispondere al messaggio OnCompleted/OnError.
Nel codice suggerisco di modificare leggermente il codice. Mi aspetto che una classe CommandReaderPublisher con un metodo Subscribe sia più appropriata di una classe CommandReader. Una volta completato un flusso Rx, non può continuare a essere utilizzato.
Inoltre mi chiedo se chiamare OnCompleted sarebbe meglio di OnError(exception) quando lo streaming sottostante si chiude. Se si verifica un errore, allora va bene, ma se si chiude allora ONCompleted potrebbe essere migliore.
+1 Non implementare le interfacce IObserver/IObservable. Usa la libreria Rx. Observable.Create/Genera/etc è il modo in cui crei una sequenza osservabile e i metodi di estensione Iscriviti sono il modo in cui il consumatore osserva una sequenza osservabile. www.IntroToRx.com. –