Code example for https://stackoverflow.com/questions/36416723/group-stream-elements-by-weight-function-in-akka-streams/36437850#36437850

WeightExample.scala

package example

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl._

object WeightExample {

  def main(args: Array[String]) {

    implicit val system = ActorSystem("mySystem")
    import system.dispatcher
    implicit val materializer = ActorMaterializer()

    type Element = String
    val weightFunction: Element => Int = _.length
    val Limit: Int = 5

    val testInput = List("One", "Two", "One", "Two", "One", "Two", "Three", ",",
      "this", "is", "a", "test", "!", "!")

    Source.fromIterator(() => testInput.toIterator)
      .scan[(Option[Element], Int)](None, 0) { case ((_, totalWeight), elem) =>
        val newWeight =
          if (totalWeight > Limit) weightFunction(elem)
          else totalWeight + weightFunction(elem)
        (Some(elem), newWeight)
      }.collect { case (Some(elem), weight) => (elem, weight > Limit) }
      .splitAfter(_._2)
      .fold(List.empty[Element]){ case (list, (elem, _)) => elem :: list}
      .map(_.reverse)
      .concatSubstreams
      .runForeach(println).onComplete(_ => system.terminate())

  }
}