Scala Akka Stream: Comment passer à travers un Seq

Question

J'essaie d'envelopper certains appels de blocage dans Future. Le type de retour est Seq[User] où User est un case class. Ce qui suit ne serait tout simplement pas compiler avec des plaintes de diverses versions surchargées étant présents. Aucune suggestion? J'ai essayé presque toutes les variations est Source.apply sans aucune chance.

// All I want is Seq[User] => Future[Seq[User]] 

def findByFirstName(firstName: String) = { 
    val users: Seq[User] = userRepository.findByFirstName(firstName) 

    val sink = Sink.fold[User, User](null)((_, elem) => elem) 

    val src = Source(users) // doesn't compile 

    src.runWith(sink) 
} 

Answer 1

Tout d'abord, je suppose que vous utilisez la version 1.0 de akka-http-experimental depuis l'API peut changé de la version précédente.

La raison pour laquelle votre code ne compile pas est que le akka.stream.scaladsl.Source$.apply() exige scala.collection.immutable.Seq au lieu de scala.collection.mutable.Seq.

Par conséquent, vous devez convertir une séquence mutable en une séquence immuable en utilisant la méthode to[T].

Document: akka.stream.scaladsl.Source

De plus, comme vous voyez le document, Source$.apply() accepte ()=>Iterator[T] que vous pouvez également passer ()=>users.iterator comme argument.

Puisque Sink.fold(...) renvoie la dernière expression évaluée, vous pouvez donner un Seq() vide comme premier argument, parcourir sur le users en ajoutant l'élément à la séquence, et enfin obtenir le résultat.

Cependant, il pourrait y avoir une meilleure solution qui peut créer un Sink qui place chaque expression évaluée dans Seq, mais je ne l'ai pas trouvé.

Le code suivant fonctionne.

import akka.actor._ 
import akka.stream.ActorMaterializer 
import akka.stream.scaladsl.{Source,Sink} 
import scala.concurrent.ExecutionContext.Implicits.global 

case class User(name:String) 

object Main extends App{ 
    implicit val system = ActorSystem("MyActorSystem") 
    implicit val materializer = ActorMaterializer() 
    val users = Seq(User("alice"),User("bob"),User("charlie")) 

    val sink = Sink.fold[Seq[User], User](Seq())(
     (seq, elem) => 
     {println(s"elem => ${elem} \t| seq => ${seq}");seq:+elem}) 

    val src = Source(users.to[scala.collection.immutable.Seq]) 
    // val src = Source(()=>users.iterator) // this also works 

    val fut = src.runWith(sink) // Future[Seq[User]] 
    fut.onSuccess({ 
     case x=>{ 
     println(s"result => ${x}") 
     } 
    }) 
} 

La sortie du code ci-dessus est

elem => User(alice)  | seq => List() 
elem => User(bob)  | seq => List(User(alice)) 
elem => User(charlie) | seq => List(User(alice), User(bob)) 
result => List(User(alice), User(bob), User(charlie)) 

Answer 2

Si vous avez juste besoin d'avenir [Seq [utilisateurs]] DonT utiliser AKKA flux mais futures

import scala.concurrent._ 
import ExecutionContext.Implicits.global 
val session = socialNetwork.createSessionFor("user", credentials) 
val f: Future[List[Friend]] = Future { 
    session.getFriends() 
}