Osservabile da Futures - onNext da più thread

Question

Voglio generare un Observable in tempo reale dai risultati di un elenco di Futures.

Nel caso più semplice, si supponga di avere un elenco di futuri che sto utilizzando con Future.sequence e sto semplicemente monitorando i loro progressi con un Observable che mi dice ogni volta che uno ha completato. Fondamentalmente lo faccio in questo modo:

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

Questo funziona perfettamente nel mio ambiente di test. Ma ho appena letto che onNext non dovrebbe essere chiamato da thread diversi, almeno senza fare attenzione che non ci siano chiamate sovrapposte. Qual è il modo consigliato per risolvere questo problema? Sembra che molti usi reali di Observables richiederebbero il onNext per essere chiamati dal codice asincrono come questo, ma non riesco a trovare un esempio simile nei documenti.

Answer 1

The Observable Contract

osservabili rilascia le notifiche agli osservatori in serie (non in parallelo). Possono emettere queste notifiche da diversi thread, ma deve esserci una relazione formale prima-avvenuta tra le notifiche .

Date un'occhiata a Serialize

E 'possibile per un' osservabile per richiamare i metodi propri osservatori in modo asincrono, forse da diversi thread. Ciò potrebbe rendere tale un Observable violento il contratto Observable, in quanto potrebbe provare a inviare una notifica OnCompleted o OnError prima di una delle sue notifiche OnNext oppure potrebbe eseguire una notifica OnNext da due thread contemporaneamente. È possibile forzare tale Osservabile a essere ben educato e sincrono applicando l'operatore Serialize su di esso.