Akka Streams:如何按大小对源中的文件列表进行分组？

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{FileIO, Flow, Sink, Source}
import akka.util.ByteString
import java.nio.file.Paths

object FileGroupingExample extends App {
  implicit val system = ActorSystem("FileGroupingExample")
  implicit val materializer = ActorMaterializer()

  // 源文件列表
  val fileList = List("file1.txt", "file2.txt", "file3.txt", "file4.txt")

  // 按大小分组的阈值（字节数）
  val groupSizeThreshold = 1000000

  // 读取文件并计算文件大小
  val fileSource = Source(fileList)
    .map(file => (file, Paths.get(file)))
    .mapAsync(parallelism = 1) { case (file, path) =>
      FileIO.fromPath(path)
        .runFold(0L)((size, bytes) => size + bytes.length)
        .map(size => (file, size))
    }

  // 根据文件大小进行分组
  val groupBySizeFlow = Flow[(String, Long)]
    .groupBy(fileSize => if (fileSize._2 > groupSizeThreshold) "large" else "small")
    .fold(List.empty[(String, Long)])((acc, fileSize) => fileSize :: acc)
    .mergeSubstreams

  // 打印分组结果
  val printSink = Sink.foreach[List[(String, Long)]](group => println(s"Group: $group"))

  // 运行流处理
  fileSource.via(groupBySizeFlow).runWith(printSink)
}