来自ftp的Akka流，逐行

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

import scala.concurrent.Future

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

  // FTP服务器地址
  val ftpServer = "ftp://ftp.example.com"
  // FTP文件路径
  val filePath = "/path/to/file.txt"

  val ftpSource: Source[ByteString, Future[IOResult]] = FileIO.fromPath(Paths.get(ftpServer + filePath))

  val processLine: Flow[ByteString, String, NotUsed] = Framing.delimiter(ByteString("\n"), maximumFrameLength = 256, allowTruncation = true)
    .map(_.utf8String)

  val printSink: Sink[String, Future[Done]] = Sink.foreach(println)

  val ftpAkkaStream: RunnableGraph[Future[Done]] = ftpSource.via(processLine).to(printSink)

  ftpAkkaStream.run().onComplete { _ =>
    system.terminate()
  }
}