Conversione di un metodo asincrono per tornare IObservable <>

Question

Ho un metodo di async che è un metodo di lunga durata che legge un ruscello e quando trova qualcosa genera un evento:

public static async void GetStream(int id, CancellationToken token) 

Si prende un gettone cancellazione perché è creato in un nuovo compito. Internamente chiama await quando legge un flusso:

var result = await sr.ReadLineAsync() 

Ora, voglio convertire questo a un metodo che restituisce un IObservable <> in modo che posso usare questo con le estensioni reattivi. Da quello che ho letto, il modo migliore per farlo è usare Observable.Create, e dal momento che RX 2.0 ora supporta anche asincrona posso ottenere tutto per lavorare con qualcosa di simile:

public static IObservable<Message> ObservableStream(int id, CancellationToken token) 
{ 
    return Observable.Create<Message>(
     async (IObserver<Message> observer) => 
      { 

Il resto del codice all'interno è lo stesso, ma invece di lanciare eventi, sto chiamando observer.OnNext(). Ma questo sembra sbagliato. Per prima cosa sto mixando CancellationTokens e, sebbene l'aggiunta della parola chiave async abbia funzionato, è questa la cosa migliore da fare? Chiamo il mio ObservableStream in questo modo:

Client.ObservableStream(555404, token).ObserveOn(Dispatcher.CurrentDispatcher).SubscribeOn(TaskPoolScheduler.Default).Subscribe(m => Messages.Add(m)); 

Answer 1

si dovrebbe cambiare GetStream per restituire un compito, invece di void (ritorno asincrona vuoto non è buona, tranne quando assolutamente necessario, come svick commentato). Una volta restituita un'attività, puoi semplicemente chiamare. ToObservable() e il gioco è fatto.

Ad esempio:

public static async Task<int> GetStream(int id, CancellationToken token) { ... } 

Poi,

GetStream(1, new CancellationToken(false)) 
    .ToObservable() 
    .Subscribe(Console.Write); 

Answer 2

Lei ha ragione. Una volta che si rappresenta l'interfaccia tramite IObservable, è necessario evitare di richiedere ai chiamanti di fornire uno CancellationToken. Ciò non significa che non puoi usarli internamente. Rx fornisce diversi meccanismi per produrre le istanze CancellationToken che vengono cancellate quando l'osservatore si annulla dal tuo osservabile.

Ci sono diversi modi per affrontare il problema. Il più semplice non richiede quasi modifiche nel codice. Esso utilizza un sovraccarico di Observable.Create cui si fornisce una CancellationToken che fa scattare, se le si cancella chiamante: non

public static IObservable<Message> ObservableStream(int id) 
{ 
    return Observable.Create<Message>(async (observer, token) => 
    { 
     // no exception handling required. If this method throws, 
     // Rx will catch it and call observer.OnError() for us. 
     using (var stream = /*...open your stream...*/) 
     { 
      string msg; 
      while ((msg = await stream.ReadLineAsync()) != null) 
      { 
       if (token.IsCancellationRequested) { return; } 
       observer.OnNext(msg); 
      } 
      observer.OnCompleted(); 
     } 
    }); 
}