buffering and interleaving results

SlickContactDAO.scala

  def streamByUserId(
    userId: UserId,
    meta:   QueryMeta
  ): Future[StreamQueryResult[Contact]] = {
    createFilteredStreamResult(queryByUserId(userId, Some(meta)), meta).map { result =>
      var buffer: Seq[((ContactDTO, ContactPointDTO), ContactTypeDTO)] = Seq.empty
      val reduced: Source[Contact, _] = result.stream.map { x =>
        buffer.headOption match {
          case Some(y) if y._1._1 == x._1._1 =>
            buffer = buffer :+ x
            None
          case Some(y) =>
            val contact = combine(buffer)
            buffer = Seq(x)
            contact
          case None =>
            buffer = Seq(x)
            None
        }
      }.filter(_.isDefined).map(_.get)
      val remainder: Source[Contact, _] = Source(() => combine(buffer).toIterator)
      val combined = reduced ++ remainder
      StreamQueryResult[Contact](result.totalCount, combined)
    }
  }