I've recently been learning to work with Akka Streams in my free time (both in Scala and java) and was wondering how to implement the following scenario.
I have a continuous stream of very large Collections coming into my pipeline and I would like to let the pipeline transform the elements inside each Collection.
Transforming a Collection into a stream of its elements is easy but I also need to collect all of the transformed elements from 1 Collection back together into 1 new Collection (containing only the transformed objects that were previously also together in the original collection). So I would have to know when a specific stream of elements of 1 Collection has been processed because then I can emit that transformed collection for further handling in the general pipeline.
As suggested by commenters, you can use fold in your transformationPipeline to assemble the List-type elements. To maintain the List boundaries when running the Stream, instead of mapConcat use flatMapConcat, as shown in the following trivialized example:
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)