Akka（39）： Http：File streaming-文件交換

  所謂文件交換指的是Http協議中服務端和客戶端之間文件的上傳和下載。Akka-http作為一種系統集成工具應該具備高效率的數據交換方式包括文件交換和數據庫表行的上傳下載。Akka-http的數據交換模式支持流式操作：代表交換數據可以是一種無限長度流的元素。這種模式首先解決了純Http大數據通過Multipart傳輸所必須進行的數據分段操作和復雜的消息屬性設定等需要的技術門檻，再者用戶還可以很方便的使用Akka-stream對數據進行深度處理，免去了數據轉換的麻煩。更重要的是：Akka-http還支持reactive-stream，可以避免由傳輸速率所產生的種種問題。在本篇我們討論利用Akka-http進行文件的雙向傳遞。

 任何文件的內容儲存格式無論在硬盤、內存或者數據線上都是一堆bytes。文件交換流程包括讀取文件里的bytes，傳送這些bytes，最終把這些bytes寫入文件。我們看到這里每個環節操作目標都是bytes，所以可能在程序里是不需要任何數據轉換過程的。Akka提供了一組文件讀寫函數，如下：

  def fromPath(f: Path, chunkSize: Int = 8192): Source[ByteString, Future[IOResult]] = fromPath(f, chunkSize, startPosition = 0) def fromPath(f: Path, chunkSize: Int, startPosition: Long): Source[ByteString, Future[IOResult]] = Source.fromGraph(new FileSource(f, chunkSize, startPosition, DefaultAttributes.fileSource, sourceShape("FileSource"))) def toPath(f: Path, options: Set[OpenOption] = Set(WRITE, TRUNCATE_EXISTING, CREATE)): Sink[ByteString, Future[IOResult]] = toPath(f, options, startPosition = 0) def toPath(f: Path, options: Set[OpenOption], startPosition: Long): Sink[ByteString, Future[IOResult]] = Sink.fromGraph(new FileSink(f, startPosition, options, DefaultAttributes.fileSink, sinkShape("FileSink")))

我們看到：fromPath類型是Source[ByteSgtring,_]，toPath類型是Sink[ByteString,_]，直接就是流型式，應該可以直接放入Http消息的Entity中，如下： 

  def fileStream(filePath: String, chunkSize: Int): Source[ByteString,Any] = { def loadFile = { // implicit val ec = httpSys.dispatchers.lookup("akka.http.blocking-ops-dispatcher")
      val file = Paths.get(filePath) FileIO.fromPath(file, chunkSize) .withAttributes(ActorAttributes.dispatcher("akka.http.blocking-ops-dispatcher")) } limitableByteSource(loadFile) }

fileStream是Source[ByteString,_]可以直接放進Entity：

  val uploadText = HttpRequest(HttpMethods.POST,uri = s"http://localhost:8011/file/text") val textData = HttpEntity( ContentTypes.`application/octet-stream`, fileStream("/Users/tiger-macpro/downloads/A4.TIF",256) )

我們把fileStream放入了HttpRequest中。對於HttpResponse可以用下面的方式：

 val route = pathPrefix("file") { (get & path("text" / Remaining)) { fp => withoutSizeLimit { complete( HttpEntity( ContentTypes.`application/octet-stream`, fileStream("/users/tiger-macpro/" + fp, 256)) ) }

注意：complete進行了HttpResponse的構建。因為Entity.dataByes就是Source[ByteString,_]，所以我們可以直接把它導入Sink：

          entity.dataBytes.runWith(FileIO.toPath(Paths.get(destPath))) .onComplete { case _ => println(s"Download file saved to: $destPath") }

上面我們提過FileIO.toPath就是一個Sink。由於我們的目的是大型的文件交換，所以無論上傳下載都使用了withoutSizeLimit：

 val route = pathPrefix("file") { (get & path("exchange" / Remaining)) { fp => withoutSizeLimit { complete( HttpEntity( ContentTypes.`application/octet-stream`, fileStream("/users/tiger-macpro/" + fp, 256)) ) } } ~ (post & path("exchange")) { withoutSizeLimit { extractDataBytes { bytes => val fut = bytes.runWith(FileIO.toPath(Paths.get(destPath))) onComplete(fut) { _ => complete(s"Save upload file to: $destPath") } } } }  

好了下面的示范代碼里對字符型或二進制文件都進行了交換的示范操作：

服務端：

 

import akka.actor._ import akka.stream._ import akka.stream.scaladsl._ import akka.http.scaladsl.Http import akka.http.scaladsl.server.Directives._ import akka.http.scaladsl.model._ import akka.http.scaladsl.model.HttpEntity._ import java.nio.file._ object FileServer extends App { implicit val httpSys = ActorSystem("httpSystem") implicit val httpMat = ActorMaterializer() implicit val httpEC = httpSys.dispatcher def fileStream(filePath: String, chunkSize: Int) = { def loadFile = { // implicit val ec = httpSys.dispatchers.lookup("akka.http.blocking-ops-dispatcher")
       val file = Paths.get(filePath) FileIO.fromPath(file, chunkSize) .withAttributes(ActorAttributes.dispatcher("akka.http.blocking-ops-dispatcher")) } limitableByteSource(loadFile) } val destPath = "/users/tiger-macpro/downloads/A4-1.TIF" val route = pathPrefix("file") { (get & path("exchange" / Remaining)) { fp => withoutSizeLimit { complete( HttpEntity( ContentTypes.`application/octet-stream`, fileStream("/users/tiger-macpro/" + fp, 256)) ) } } ~ (post & path("exchange")) { withoutSizeLimit { extractDataBytes { bytes => val fut = bytes.runWith(FileIO.toPath(Paths.get(destPath))) onComplete(fut) { _ => complete(s"Save upload file to: $destPath") } } } } } val (port, host) = (8011,"localhost") val bindingFuture = Http().bindAndHandle(route,host,port) println(s"Server running at $host $port. Press any key to exit ...") scala.io.StdIn.readLine() bindingFuture.flatMap(_.unbind()) .onComplete(_ => httpSys.terminate()) }

 

客戶端：

 

import akka.actor._ import akka.stream._ import akka.stream.scaladsl._ import akka.http.scaladsl.Http import akka.http.scaladsl.model.HttpEntity.limitableByteSource import akka.http.scaladsl.model._ import java.nio.file._ import akka.util.ByteString import scala.util._ object FileClient extends App { implicit val sys = ActorSystem("ClientSys") implicit val mat = ActorMaterializer() implicit val ec = sys.dispatcher def downloadFileTo(request: HttpRequest, destPath: String) = { val futResp = Http(sys).singleRequest(request) futResp .andThen { case Success(r@HttpResponse(StatusCodes.OK, _, entity, _)) => entity.dataBytes.runWith(FileIO.toPath(Paths.get(destPath))) .onComplete { case _ => println(s"Download file saved to: $destPath") } case Success(r@HttpResponse(code, _, _, _)) => println(s"Download request failed, response code: $code") r.discardEntityBytes() case Success(_) => println("Unable to download file!") case Failure(err) => println(s"Download failed: ${err.getMessage}") } } val dlFile = "Downloads/readme.txt" val downloadText = HttpRequest(uri = s"http://localhost:8011/file/exchange/" + dlFile) downloadFileTo(downloadText, "/users/tiger-macpro/downloads/sample.txt") scala.io.StdIn.readLine() val dlFile2 = "Downloads/image.png" val downloadText2 = HttpRequest(uri = s"http://localhost:8011/file/exchange/" + dlFile2) downloadFileTo(downloadText2, "/users/tiger-macpro/downloads/sample.png") scala.io.StdIn.readLine() def uploadFile(request: HttpRequest, dataEntity: RequestEntity) = { val futResp = Http(sys).singleRequest( request.copy(entity = dataEntity) ) futResp .andThen { case Success(r@HttpResponse(StatusCodes.OK, _, entity, _)) => entity.dataBytes.map(_.utf8String).runForeach(println) case Success(r@HttpResponse(code, _, _, _)) => println(s"Upload request failed, response code: $code") r.discardEntityBytes() case Success(_) => println("Unable to Upload file!") case Failure(err) => println(s"Upload failed: ${err.getMessage}") } } def fileStream(filePath: String, chunkSize: Int): Source[ByteString,Any] = { def loadFile = { // implicit val ec = httpSys.dispatchers.lookup("akka.http.blocking-ops-dispatcher")
      val file = Paths.get(filePath) FileIO.fromPath(file, chunkSize) .withAttributes(ActorAttributes.dispatcher("akka.http.blocking-ops-dispatcher")) } limitableByteSource(loadFile) } val uploadText = HttpRequest(HttpMethods.POST,uri = s"http://localhost:8011/file/exchange") val textData = HttpEntity( ContentTypes.`application/octet-stream`, fileStream("/Users/tiger-macpro/downloads/readme.txt",256) ) uploadFile(uploadText,textData) scala.io.StdIn.readLine() sys.terminate() }