Immutable stream batching

Viewed 90

Is there an immutable alternative to the solution in this question, which batches up data in a stream:

val records =
  Source(List(
    Record(1, "a"),
    Record(1, "k"),
    Record(1, "k"),
    Record(1, "a"),
    Record(2, "r"),
    Record(2, "o"),
    Record(2, "c"),
    Record(2, "k"),
    Record(2, "s"),
    Record(3, "!")
  ))
  .concat(Source.single(Record(0, "notused"))) // needed to print the last element

records
  .statefulMapConcat { () =>
    var currentTime = 0
    var payloads: Seq[String] = Nil

    record =>
      if (record.time == currentTime) {
        payloads = payloads :+ record.payload
        Nil
      } else {
        val previousState = (currentTime, payloads)
        currentTime = record.time
        payloads = Seq(record.payload)
        List(previousState)
      }
  }
  .runForeach(println)

Produces

(0,List())
(1,List(a, k, k, a))
(2,List(r, o, c, k, s))
(3,List(!))
1 Answers

you're on a good way, immutability and stateless is a important aspect in programming concurrent distributed software. I share this scastie with your sample code while using groupBy from akka. Let me know, if it will help you. The output is like

(3,List(!))
(1,List(a, k, k, a))
(2,List(r, o, c, k, s))
Related