TraversableOnce, Future e Option in a Scala per comprensione

Question

Ho una lista di identificativi di stringa che rappresentano i record DB. Mi piacerebbe caricarli dal DB in modo asincrono, quindi caricare ogni record su un server remoto in modo asincrono, quindi quando tutti hanno finito di caricare, registra gli id ​​dei record che sono stati caricati.

Dal momento che sono su Scala 2.9.2, sto utilizzando l'implementazione Future di nucleo-util di Twitter, ma dovrebbe funzionare esattamente come i futuri 2.10 in termini di trasformazioni monadiche.

Il concetto generale è questo:

def fetch(id: String): Future[Option[Record]] 
def upload(record: Record): Future[String] 
def notifyUploaded(ids: Seq[String]): Unit 

val ids: Seq[String] = .... 

che sto cercando di fare questo attraverso una per la comprensione ma il fatto che recuperare restituisce un futuro di opzione rende oscuro e il codice non viene compilato:

for { 
    id <- ids 
    maybeRecord <- fetch(id) 
    record <- maybeRecord 
    uploadedId <- upload(record) 
} yield uploadedId 

compilazione il risultato è il seguente errore:

scala: type mismatch; 
found : com.twitter.util.Future[String] 
required: Option[?] 
    uploadedId <- upload(record) 
       ^

Cosa sto mI ssing? perché il compilatore si aspetta che uploadId sia un'opzione? c'è un modo carino per aggirare questo?

Answer 1

consideri la firma della funzione flatMap (o bind):

trait Monad[M[_]] { 
    def flatMap[A](a : M[A], f : A => M[B]) : M[B] 
    .... 

Nel tuo caso, si sta cercando di utilizzare flatMap su un Option, dandogli un f che genera un Future. Ma come nella firma sopra, f dovrebbe generare qualcosa nella stessa monade su cui è stato chiamato.

Scala non è necessariamente terribilmente utile a questo proposito, dal momento che è abbastanza bravo a convertire le cose (per Seq s, per esempio) in modo tale che si ha l'impressione che si può catena arbitrarie flatMap chiamate insieme, a prescindere del contenitore.

Quello che si potrebbe desiderare è un 'trasformatore Monade', che ti dà la possibilità di comporre le monadi. Debasish Ghosh ha un post sull'uso dei trasformatori Scalad Monad here.

Answer 2

Non è possibile combinare tutti i diversi tipi in uno per la comprensione, ho capito che si potrebbe mescolare Seq e Opzione e il risultato sarebbe Seq o Opzione a seconda di ciò che è prima. Non è possibile combinare Future e Seq o Option. Se vuoi usare per la comprensione, dovresti ridurli a pochi. In questi casi potrebbe essere più bello con map/flatMap. Ho implementato la tua domanda in entrambi i modi e ho aggiunto i tipi a pochi risultati intermedi in modo da poter vedere il disordine che si sta creando mentre si lavora con tutti quei tipi diversi.

object TestClass { 

    import scala.concurrent.Future 
    import scala.concurrent.ExecutionContext.Implicits.global 
    import scala.concurrent._ 
    import scala.concurrent.duration._ 

    case class Record(id: String) 


    def fetch(id: String): Future[Option[Record]] = Future { 
    Thread.sleep(1000); 
    Some(Record(id)) 
    } 

    def upload(record: Record): Future[String] = Future { 
    Thread.sleep(3000); 
    record.id + "_uploaded" 
    } 

    def notifyUploaded(ids: Seq[String]): Unit = println("notified" + ids) 

    val ids: Seq[String] = Seq("a", "b", "c") 

    def main(args: Array[String]): Unit = { 
    forComprehensionImpl() 
    mapAndFlatMapImpl() 
    } 

    def forComprehensionImpl() = { 
    val result: Seq[Future[Option[Future[String]]]] = for { 
     id <- ids 
    } yield { 
     for { 
     maybeRecord <- fetch(id) 
     } yield { 
     for { 
      record <- maybeRecord 
     } yield { 
      for { 
      uploadedId <- upload(record) 
      } yield uploadedId 
     } 
     } 
    } 
    val result2: Future[Seq[Option[Future[String]]]] = Future.sequence(result) 
    val result3: Future[Unit] = result2.flatMap { x: Seq[Option[Future[String]]] => 
     Future.sequence(x.flatten).map(notifyUploaded) 
    } 
    Await.result(result3, Duration.Inf) 
    } 


    def mapAndFlatMapImpl() = { 
    val res: Seq[Future[Iterable[String]]] = ids.map { id => 
     fetch(id).flatMap { maybeRecord => 
     val res1: Option[Future[Seq[String]]] = maybeRecord.map { record => 
      upload(record) map (Seq(_)) 
     } 
     res1 match { 
      case Some(a) => a 
      case None => Future(Seq()) 
     } 
     } 
    } 
    val res3: Future[Unit] = Future.sequence(res) map (a => notifyUploaded(a.flatten)) 
    Await.result(res3, Duration.Inf) 
    } 
}