Konwersja metody asynchronicznej wrócić IObservable <>

Question

Mam async metodę, która jest długo działa metoda, która czyta strumień i gdy znajdzie coś uruchamia zdarzenie:

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

Zajmuje token anulowania ponieważ to jest tworzenie w nowym zadaniu. Wewnętrznie nazywa to await gdy czyta strumień:

var result = await sr.ReadLineAsync() 

Teraz chcę przekonwertować to do metody, która zwraca IObservable <> tak, że mogę używać tego z reaktywnymi rozszerzeń. Z tego co czytałem, najlepszym sposobem na to jest za pomocą Observable.Create i od RX 2.0 obsługuje teraz także asynchroniczny mogę to wszystko do pracy z mniej więcej tak:

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

reszty kodu wewnątrz jest taki sam, ale zamiast wystrzeliwania zdarzeń dzwonię pod numer observer.OnNext(). Ale to wydaje się złe. Z jednej strony miksuję tam CancellationTokeny, i chociaż dodanie słowa asynchronicznego sprawiło, że działa, czy to naprawdę najlepsza rzecz do zrobienia? Dzwonię do mojego ObservableStream tak:

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

Answer 1

należy zmienić GetStream zwrócić zadanie, zamiast pustki (powrót asynchronicznej pustkę nie jest dobre, z wyjątkiem gdy jest to absolutnie konieczne, ponieważ svick skomentował). Gdy zwrócisz zadanie, możesz po prostu zadzwonić .ToObservable() i gotowe.

Na przykład:

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

Następnie

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

Answer 2

Jesteś poprawne. Po przedstawieniu interfejsu za pomocą interfejsu IObservable należy unikać wymagania od dzwoniących do dostarczenia numeru CancellationToken. To nie znaczy, że nie możesz ich używać wewnętrznie. Rx udostępnia kilka mechanizmów do tworzenia instancji CancellationToken, które są anulowane, gdy obserwator zrezygnuje z obserwowania.

Istnieje wiele sposobów na rozwiązanie problemu. Najprostszy wymaga prawie żadnych zmian w kodzie. Wykorzystuje przeciążenie Observable.Create który dostarcza Ci CancellationToken skutkującym jeśli unsubscribes dzwoniących:

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