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()


}

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Hongten

Java Web 网络留言板3 Commons-DbUtils

<html> <head> <meta http-equiv="Content-Type" content="text/html; charset=utf-8"...

15020
来自专栏微信公众号:Java团长

Java POI 导出EXCEL经典实现

在web开发中,有一个经典的功能,就是数据的导入导出。特别是数据的导出,在生产管理或者财务系统中用的非常普遍,因为这些系统经常要做一些报表打印的工作。而数据导出...

72620
来自专栏lgp20151222

filter和listener的生命周期

17050
来自专栏区块链技术专栏

EOSIO 转帐详解

在EOS网络中存在两种货币,一种是EOS,还有一种是EOS网络中的代币。说到这里大家似乎有点疑惑,举个简单的例子,就好比ETH网络中的ETH,ETH网络中的其他...

79670
来自专栏海说

10、借助POI实现Java生成并打印excel报表(1)

10.1、了解 Apache POI 实际开发中,用到最多的是把数据库中数据导出生成报表,尤其是在生产管理或者财务系统中用的非常普遍。生成报表格式一般是EXC...

99700
来自专栏lgp20151222

java对redis的操作

16810
来自专栏Hongten

Java Web 网络留言板8

  admin:id,name ,password                  <pk>id

27110
来自专栏GIS讲堂

大量POI的解决方案2

在前面的文章中,讲述了通过“抽稀+后台生成图片”的方式解决大量POI点展示的一种思路,后面看了tilestache的矢量切片方式,自己仔细思考了下,提出了本文大...

20150
来自专栏别先生

统计各个数据库的各个数据表的总数,然后写入到excel中

1、最近项目基本进入最后阶段了,然后会统计一下各个数据库的各个数据表的数据量,开始使用的报表工具,report-designer,开源的,研究了两天,发现并不是...

31720
来自专栏GIS讲堂

Java中用Apache POI生成excel和word文档

最近在做项目的过程中遇到了excel的数据导出和word的图文表报告的导出功能,最后决定用Apache POI来完成该项功能。本文就项目实现过程中的一些思路与代...

40840

扫码关注云+社区

领取腾讯云代金券