Transforming the inner elements of a stream of collections

user_14653519 0
Viewed 80

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.

akka-stream scala java
1 Answers

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)
user_63165080
Related