Observable from Futures - onNastępnie z wielu wątków

Question

Chcę wygenerować Observable w czasie rzeczywistym z wyników listy Futures.

W najprostszym przypadku, załóżmy, że mam listę kontraktów terminowych, które prowadzę z Future.sequence, i po prostu monitoruję ich postępy za pomocą Observable, która mówi mi za każdym razem, gdy się zakończy. Zasadniczo robię to tak:

def observeFuturesProgress(futs: List[Future[Int]]) : Observable[String] = { 
    Observable[String](observer => { 
     val loudFutures: List[Future[Int]] = futs.map(f => { 
      f onComplete { 
       case Success(a) => observer.onNext(s"just did $a more") 
       case Failure(e) => observer.onError(e) 
      } 
      f 
     }) 
     Future.sequence(loudFutures) onComplete { 
      case Success(_) => observer.onCompleted() 
      case Failure(e) => observer.onError(e) 
     } 
    }) 
    } 

To działa dobrze w moim środowisku testowym. Ale właśnie przeczytałem, że onNext nie powinien być wywoływany z różnych wątków, przynajmniej bez bycia ostrożnym, że nie ma nakładających się połączeń. Jaki jest zalecany sposób rozwiązania tego problemu? Wydaje się, że wiele rzeczywistych zastosowań Observables wymagałoby wywołania onNext z takiego kodu asynchronicznego, ale nie mogę znaleźć podobnego przykładu w dokumentach.

Answer 1

The Observable Contract

obserwable musi wydać powiadomienia obserwatorów seryjnie (nie w równolegle). Mogą wysyłać te powiadomienia z różnych wątków, , ale musi istnieć formalna, zdarzająca się wcześniej relacja między powiadomieniami .

Spójrz na Serialize

Jest to możliwe do zaobserwowania w celu wywołania metod swoich obserwatorów asynchronicznie, być może z różnych wątków. Może to spowodować, że obserwowalne naruszy kontrakt obserwowalny, ponieważ może próbować wysłać do powiadomienie OnCompleted lub OnError przed jednym z powiadomień OnNext lub może jednocześnie wysyłać powiadomienia OnNext z dwóch różnych wątków . Możesz wymusić takie obserwowanie, aby być dobrze zachowanym i synchronicznym, stosując do tego operator szeregowy.