Recientemente he estado aprendiendo a trabajar con Akka Streams en mi tiempo libre (tanto en Scala como en Java) y me preguntaba cómo implementar el siguiente escenario.
Tengo un flujo continuo de colecciones muy grandes que ingresan a mi canalización y me gustaría dejar que la canalización transforme los elementos dentro de cada colección.
Transformar una colección en un flujo de sus elementos es fácil, pero también necesito reunir todos los elementos transformados de 1 colección en 1 nueva colección (que contenga solo los objetos transformados que anteriormente también estaban juntos en la colección original). Por lo tanto, tendría que saber cuándo se procesó un flujo específico de elementos de 1 Colección porque entonces puedo emitir esa colección transformada para su posterior manejo en la canalización general.
Como lo sugieren los comentaristas, puede usar fold en su transformationPipeline para ensamblar los elementos de tipo Lista. Para mantener los límites de la Lista cuando se ejecuta el Stream, en lugar de mapConcat , use flatMapConcat , como se muestra en el siguiente ejemplo trivializado:
def transform(s: String): Int = s.length val transformationPipeline: Flow[String, List[Int], NotUsed] = Flow[String]. fold(List.empty[Int])((ls, s) => transform(s) :: ls). map(_.reverse) val flow: Flow[List[String], List[Int], NotUsed] = Flow[List[String]]. flatMapConcat(Source(_).via(transformationPipeline)) Source(List("a", "bb") :: List("cc", "ddd", "e") :: Nil). via(flow). runForeach(println) // List(1, 2) // List(2, 3, 1)