上篇我们谈论了Akka-http的文件交流。由于文件内容编码和传导线上数据表明型式皆为bytes,所以可以直接把公文内容存进HttpEntity中开展传递。那么对于在内存里自定义的高档数据类型则应当需要首先举行byte转换后才能放入HttpEntity中了。高级数据类型与byte之间的交互转换就是marshalling和unmarshalling过程了。这么些我们在前几篇商讨里提及过,在本篇再反复加强影像。因为大家的重要性目标是落实数据库表行的置换,所以应该把热点放在
T <->
MessageEntity这样的更换上。在Akka-http中T->MessageEntity转换是经过马尔斯(Mars)haller[T,MessageEntity]心想事成的,马尔斯(Mars)haller类型定义如下:

sealed abstract class Marshaller[-A, +B] {

  def apply(value: A)(implicit ec: ExecutionContext): Future[List[Marshalling[B]]]
...
}
object Marshaller
  extends GenericMarshallers
  with PredefinedToEntityMarshallers
  with PredefinedToResponseMarshallers
  with PredefinedToRequestMarshallers {

  /**
   * Creates a [[Marshaller]] from the given function.
   */
  def apply[A, B](f: ExecutionContext ⇒ A ⇒ Future[List[Marshalling[B]]]): Marshaller[A, B] =
    new Marshaller[A, B] {
      def apply(value: A)(implicit ec: ExecutionContext) =
        try f(ec)(value)
        catch { case NonFatal(e) ⇒ FastFuture.failed(e) }
    }
...

其一连串包嵌了个类型转换函数:A
=>
Future[List[Marshalling[B]]],最后目的是A=>B的变换。扩张了一层马尔斯halling类型是为着更有利对B类型目标展开筛选、修改操作。我们看看类型马尔斯hal的更换函数to[???]就精晓了:

class Marshal[A](val value: A) {
  /**
   * Marshals `value` using the first available [[Marshalling]] for `A` and `B` provided by the given [[Marshaller]].
   * If the marshalling is flexible with regard to the used charset `UTF-8` is chosen.
   */
  def to[B](implicit m: Marshaller[A, B], ec: ExecutionContext): Future[B] =
    m(value).fast.map {
      _.head match {
        case Marshalling.WithFixedContentType(_, marshal) ⇒ marshal()
        case Marshalling.WithOpenCharset(_, marshal)      ⇒ marshal(HttpCharsets.`UTF-8`)
        case Marshalling.Opaque(marshal)                  ⇒ marshal()
      }
    }

第一,在可视域内亟待马尔斯(Mars)haller[A,B]隐式实例存在,马尔斯(Mars)halling提供筛选,最终马尔斯(Mars)haller的包嵌函数marshal进行了实际的类型转换。Akka-http提供了根基数据类型到MessageEntity转换的隐式实例,如下:

trait PredefinedToEntityMarshallers extends MultipartMarshallers {

  implicit val ByteArrayMarshaller: ToEntityMarshaller[Array[Byte]] = byteArrayMarshaller(`application/octet-stream`)
  def byteArrayMarshaller(contentType: ContentType): ToEntityMarshaller[Array[Byte]] =
    Marshaller.withFixedContentType(contentType) { bytes ⇒ HttpEntity(contentType, bytes) }

  implicit val ByteStringMarshaller: ToEntityMarshaller[ByteString] = byteStringMarshaller(`application/octet-stream`)
  def byteStringMarshaller(contentType: ContentType): ToEntityMarshaller[ByteString] =
    Marshaller.withFixedContentType(contentType) { bytes ⇒ HttpEntity(contentType, bytes) }

  implicit val CharArrayMarshaller: ToEntityMarshaller[Array[Char]] = charArrayMarshaller(`text/plain`)
  def charArrayMarshaller(mediaType: MediaType.WithOpenCharset): ToEntityMarshaller[Array[Char]] =
    Marshaller.withOpenCharset(mediaType) { (value, charset) ⇒ marshalCharArray(value, mediaType withCharset charset) }
  def charArrayMarshaller(mediaType: MediaType.WithFixedCharset): ToEntityMarshaller[Array[Char]] =
    Marshaller.withFixedContentType(mediaType) { value ⇒ marshalCharArray(value, mediaType) }

  private def marshalCharArray(value: Array[Char], contentType: ContentType.NonBinary): HttpEntity.Strict =
    if (value.length > 0) {
      val charBuffer = CharBuffer.wrap(value)
      val byteBuffer = contentType.charset.nioCharset.encode(charBuffer)
      val array = new Array[Byte](byteBuffer.remaining())
      byteBuffer.get(array)
      HttpEntity(contentType, array)
    } else HttpEntity.Empty

  implicit val DoneMarshaller: ToEntityMarshaller[akka.Done] =
    Marshaller.withFixedContentType(`text/plain(UTF-8)`) { done ⇒
      HttpEntity(`text/plain(UTF-8)`, "")
    }

  implicit val StringMarshaller: ToEntityMarshaller[String] = stringMarshaller(`text/plain`)
  def stringMarshaller(mediaType: MediaType.WithOpenCharset): ToEntityMarshaller[String] =
    Marshaller.withOpenCharset(mediaType) { (s, cs) ⇒ HttpEntity(mediaType withCharset cs, s) }
  def stringMarshaller(mediaType: MediaType.WithFixedCharset): ToEntityMarshaller[String] =
    Marshaller.withFixedContentType(mediaType) { s ⇒ HttpEntity(mediaType, s) }

  implicit val FormDataMarshaller: ToEntityMarshaller[FormData] =
    Marshaller.withOpenCharset(`application/x-www-form-urlencoded`) { _ toEntity _ }

  implicit val MessageEntityMarshaller: ToEntityMarshaller[MessageEntity] =
    Marshaller strict { value ⇒ Marshalling.WithFixedContentType(value.contentType, () ⇒ value) }
}

object PredefinedToEntityMarshallers extends PredefinedToEntityMarshallers

专注:上边的那个转换函数类型都是ToEntity马尔斯(Mars)haller,这是一个种体系称,实际上就是马尔斯(Mars)haller[T,MessageEntity]:

  type ToEntityMarshaller[T] = Marshaller[T, MessageEntity]

从源代码上看这个马尔斯haller的隐式实例都提供了转移函数
T=>HttpEntity。这样就可以在骨子里类型转换时倘若能找到相应马尔斯haller的隐式实例就足以调用它的转换函数举办更换操作了。

前几天,只要通过import把这一个隐式实例导入可视域内就可以这么调用马尔斯(Mars)hal了:

import akka.http.scaladsl.marshalling.Marshal

  val aChars = Array[Char]('h','e','l','l','o')
  val aBytes = Array[Byte](0,1,2,3)
  val strHello = Marshal("Hello").to[MessageEntity]
  val chHello = Marshal(aChars).to[MessageEntity]
  val bt0123 = Marshal(aBytes).to[MessageEntity]

那么对于结构复杂的自定义类型又怎么呢?如下:

  case class Person(id: Int, name: String)
  val john = Person(12,"John")
  val futP = Marshal(john).to[MessageEntity]

以此futP无法通过编译,报错如下:

Error:(17, 30) could not find implicit value for parameter m: akka.http.scaladsl.marshalling.Marshaller[MarshalDemo.Person,akka.http.scaladsl.model.MessageEntity]
  val futP = Marshal(john).to[MessageEntity]

这是因为编译器compiler不可能找到马尔斯(Mars)haller[Person,MessageEntity]以此类此外隐式实例。现在自己只为Person自定义一个马尔斯haller隐式实例:

  implicit val PersonMarshaller: ToEntityMarshaller[Person] = personMarshaller(`text/plain`)
  def personMarshaller(mediaType: MediaType.WithOpenCharset): ToEntityMarshaller[Person] =
    Marshaller.withOpenCharset(mediaType) { (p, ps) ⇒ HttpEntity(mediaType withCharset ps, ByteString(p.toString)) }
  def personMarshaller(mediaType: MediaType.WithFixedCharset): ToEntityMarshaller[Person] =
    Marshaller.withFixedContentType(mediaType) { p ⇒ HttpEntity(mediaType, ByteString(p.toString)) }

本条马尔斯(Mars)haller代表的转移过程是:Person
-> Person.String ->
ByteString。中间多了层用String函数来叙述Person类型。这只是自我个人的变换格局,所以反向转换Unmarshalling也不可能不比照自己的艺术把String转回Person。实际上这种转移的盛开标准之一就是Json,大家共同遵照规范要求的表明形式举行转移操作就能落得相同的目标了。

Akka-http自带的Json解决方案用的是Spray-Json,下面大家就用Spray-Json来兑现转移:

import akka.http.scaladsl.marshallers.sprayjson._
import spray.json._

trait Formats extends SprayJsonSupport with DefaultJsonProtocol
object Converters extends Formats {
  case class Person(id: Int, name: String)
  implicit val userFormat = jsonFormat2(Person.apply)
}
...
  import Converters._
  val john = Person(12,"John")
  val futP = Marshal(john).to[MessageEntity]

现行的变换流程变成了:Person
-> Json ->
ByteString。Akka-http是透过Root詹森Format[T]来提供转换隐式实例的:

/**
 * A special JsonFormat signaling that the format produces a legal JSON root object, i.e. either a JSON array
 * or a JSON object.
 */
trait RootJsonFormat[T] extends JsonFormat[T] with RootJsonReader[T] with RootJsonWriter[T]

RootJsonFormat[T]代表T类型实例的Json转换。RootJsonFormat[T]的继承父辈包括:
/**
  * Provides the JSON deserialization and serialization for type T.
 */
trait JsonFormat[T] extends JsonReader[T] with JsonWriter[T]

/**
 * A special JsonReader capable of reading a legal JSON root object, i.e. either a JSON array or a JSON object.
 */
@implicitNotFound(msg = "Cannot find RootJsonReader or RootJsonFormat type class for ${T}")
trait RootJsonReader[T] extends JsonReader[T]

/**
 * A special JsonWriter capable of writing a legal JSON root object, i.e. either a JSON array or a JSON object.
 */
@implicitNotFound(msg = "Cannot find RootJsonWriter or RootJsonFormat type class for ${T}")
trait RootJsonWriter[T] extends JsonWriter[T]

在我们的事例里Person的马尔斯haller隐式实例是由此jsonFormat2函数到手的:

  def jsonFormat2[P1 :JF, P2 :JF, T <: Product :ClassManifest](construct: (P1, P2) => T): RootJsonFormat[T] = {
    val Array(p1, p2) = extractFieldNames(classManifest[T])
    jsonFormat(construct, p1, p2)
  }
  def jsonFormat[P1 :JF, P2 :JF, T <: Product](construct: (P1, P2) => T, fieldName1: String, fieldName2: String): RootJsonFormat[T] = new RootJsonFormat[T]{
    def write(p: T) = {
      val fields = new collection.mutable.ListBuffer[(String, JsValue)]
      fields.sizeHint(2 * 3)
      fields ++= productElement2Field[P1](fieldName1, p, 0)
      fields ++= productElement2Field[P2](fieldName2, p, 1)
      JsObject(fields: _*)
    }
    def read(value: JsValue) = {
      val p1V = fromField[P1](value, fieldName1)
      val p2V = fromField[P2](value, fieldName2)
      construct(p1V, p2V)
    }
  }

就是这么些函数重返了RootJsonFormat[T]。可以看到,效率的实际贯彻在jsonFormat函数里,在这边实现了对json数据结构的读写。jsonFormat2是在ProductFormatsInstances
trait里的,也就是ProductFormats: 

trait ProductFormats extends ProductFormatsInstances {
  this: StandardFormats =>

我们地方例子里的Formats
trait继承了DefaultJsonProtocal,这其中包括了有着json转换实例构建形式:

/**
  * Provides all the predefined JsonFormats.
 */
trait DefaultJsonProtocol
        extends BasicFormats
        with StandardFormats
        with CollectionFormats
        with ProductFormats
        with AdditionalFormats

object DefaultJsonProtocol extends DefaultJsonProtocol

再看看Root詹森Format及连锁后续情形:

/**
 * A special JsonFormat signaling that the format produces a legal JSON root object, i.e. either a JSON array
 * or a JSON object.
 */
trait RootJsonFormat[T] extends JsonFormat[T] with RootJsonReader[T] with RootJsonWriter[T]

/**
  * Provides the JSON deserialization and serialization for type T.
 */
trait JsonFormat[T] extends JsonReader[T] with JsonWriter[T]

/**
 * A special JsonReader capable of reading a legal JSON root object, i.e. either a JSON array or a JSON object.
 */
@implicitNotFound(msg = "Cannot find RootJsonReader or RootJsonFormat type class for ${T}")
trait RootJsonReader[T] extends JsonReader[T]

/**
 * A special JsonWriter capable of writing a legal JSON root object, i.e. either a JSON array or a JSON object.
 */
@implicitNotFound(msg = "Cannot find RootJsonWriter or RootJsonFormat type class for ${T}")
trait RootJsonWriter[T] extends JsonWriter[T]

下面是Spray-Json的现实贯彻:

package json {

  case class DeserializationException(msg: String, cause: Throwable = null, fieldNames: List[String] = Nil) extends RuntimeException(msg, cause)
  class SerializationException(msg: String) extends RuntimeException(msg)

  private[json] class PimpedAny[T](any: T) {
    def toJson(implicit writer: JsonWriter[T]): JsValue = writer.write(any)
  }

  private[json] class PimpedString(string: String) {
    @deprecated("deprecated in favor of parseJson", "1.2.6")
    def asJson: JsValue = parseJson
    def parseJson: JsValue = JsonParser(string)
  }
}

toJson,as杰森(Jason)分别需要JsonWriter,Json里德r的隐式实例。

从下边的探究中我们对自由结构类型的一个实例举办系列化转换有了一定驾驭。这么些类此外实例可以被是作为数据库的一条记下,通过下边啄磨的法门在服务端和客户端举办置换。这是因为SprayJsonSupport可以提供任意类T的马尔斯haller[T,MessageEntity]隐式实例。

因为大家的重要目标是兑现数据库表多行的交流,所以必须要贯彻以表作为因素数据流的数据互换,也就是说最起码能要在可视域内提供马尔斯hall[Source[T],_],MessageEnity]及Unmarshaller[MessageEntity,Source[T,_]]的隐式实例才行。在服务端大家品尝过用complete(Source[T,NotUsed])来形成HttpResponse的构建。但独立用马尔斯(Mars)hal(source).to[Source[T,NotUsed]]则编译出错。这是因为Akka-http提供的是ToResponse马尔斯haller[Source[T,M]]的隐式实例:

implicit def fromEntityStreamingSupportAndByteStringMarshaller[T, M](implicit s: EntityStreamingSupport, m: ToByteStringMarshaller[T]): ToResponseMarshaller[Source[T, M]] = {
    Marshaller[Source[T, M], HttpResponse] { implicit ec ⇒ source ⇒
      FastFuture successful {
        Marshalling.WithFixedContentType(s.contentType, () ⇒ {
          val availableMarshallingsPerElement = source.mapAsync(1) { t ⇒ m(t)(ec) }

          // TODO optimise such that we pick the optimal marshalling only once (headAndTail needed?)
          // TODO, NOTE: this is somewhat duplicated from Marshal.scala it could be made DRYer
          val bestMarshallingPerElement = availableMarshallingsPerElement mapConcat { marshallings ⇒
            // pick the Marshalling that matches our EntityStreamingSupport
            (s.contentType match {
              case best @ (_: ContentType.Binary | _: ContentType.WithFixedCharset | _: ContentType.WithMissingCharset) ⇒
                marshallings collectFirst { case Marshalling.WithFixedContentType(`best`, marshal) ⇒ marshal }

              case best @ ContentType.WithCharset(bestMT, bestCS) ⇒
                marshallings collectFirst {
                  case Marshalling.WithFixedContentType(`best`, marshal) ⇒ marshal
                  case Marshalling.WithOpenCharset(`bestMT`, marshal)    ⇒ () ⇒ marshal(bestCS)
                }
            }).toList
          }
          val marshalledElements: Source[ByteString, M] =
            bestMarshallingPerElement.map(_.apply()) // marshal!
              .via(s.framingRenderer)

          HttpResponse(entity = HttpEntity(s.contentType, marshalledElements))
        }) :: Nil
      }
    }
  }

以此complete(m =>
ToResponse马尔斯(Mars)hallable)是个magnet-pattern函数,巧妙在ToResponse马尔斯hallable参数类型:

/** Something that can later be marshalled into a response */
trait ToResponseMarshallable {
  type T
  def value: T
  implicit def marshaller: ToResponseMarshaller[T]

  def apply(request: HttpRequest)(implicit ec: ExecutionContext): Future[HttpResponse] =
    Marshal(value).toResponseFor(request)
}

object ToResponseMarshallable {
  implicit def apply[A](_value: A)(implicit _marshaller: ToResponseMarshaller[A]): ToResponseMarshallable =
    new ToResponseMarshallable {
      type T = A
      def value: T = _value
      def marshaller: ToResponseMarshaller[T] = _marshaller
    }

  implicit val marshaller: ToResponseMarshaller[ToResponseMarshallable] =
    Marshaller { implicit ec ⇒ marshallable ⇒ marshallable.marshaller(marshallable.value) }
}

magnet-pattern我们就不多谈了。但它的伴生对象中涵盖了对此外类型ToResponse马尔斯hallable的隐式实例,所以complete可以由此编译。SprayJsonSupport中倒是提供了Unmarshaller[MessageEntity,T]的隐式实例: 

  // support for as[Source[T, NotUsed]]
  implicit def sprayJsonSourceReader[T](implicit reader: RootJsonReader[T], support: EntityStreamingSupport): FromEntityUnmarshaller[Source[T, NotUsed]] =
    Unmarshaller.withMaterializer { implicit ec ⇒ implicit mat ⇒ e ⇒
      if (support.supported.matches(e.contentType)) {
        val frames = e.dataBytes.via(support.framingDecoder)
        val unmarshal = sprayJsonByteStringUnmarshaller(reader)(_)
        val unmarshallingFlow =
          if (support.unordered) Flow[ByteString].mapAsyncUnordered(support.parallelism)(unmarshal)
          else Flow[ByteString].mapAsync(support.parallelism)(unmarshal)
        val elements = frames.viaMat(unmarshallingFlow)(Keep.right)
        FastFuture.successful(elements)
      } else FastFuture.failed(Unmarshaller.UnsupportedContentTypeException(support.supported))
    }

总的看如若需要贯彻stream的双向互换,大家还非得提供马尔斯haller[Source[T,NotUsed],MessageEntity]以及Unmarshaller[MessageEntity,Source[T,NotUsed]]才行。篇幅所限,具体实现形式移到下篇研究。

下边是此次研讨的示范源代码:

import akka.actor._
import akka.stream.scaladsl._
import akka.http.scaladsl.marshalling._
import akka.http.scaladsl.model._
import akka.http.scaladsl.model.MediaTypes._
import akka.util.ByteString
import akka.http.scaladsl.marshallers.sprayjson._
import spray.json._


trait Formats extends SprayJsonSupport with DefaultJsonProtocol
object sprayConverters extends Formats {
  case class Person(id: Int, name: String)
  implicit val userFormat = jsonFormat2(Person.apply)
}


object MarshalDemo extends App {
  import sprayConverters._

  implicit val sys = ActorSystem("marshaller")
  implicit val ec = sys.dispatcher

  val aChars = Array[Char]('h','e','l','l','o')
  val aBytes = Array[Byte](0,1,2,3)
  val strHello = Marshal("Hello").to[MessageEntity]
  val chHello = Marshal(aChars).to[MessageEntity]
  val bt0123 = Marshal(aBytes).to[MessageEntity]

  implicit val PersonMarshaller: ToEntityMarshaller[Person] = personMarshaller(`text/plain`)
  def personMarshaller(mediaType: MediaType.WithOpenCharset): ToEntityMarshaller[Person] =
    Marshaller.withOpenCharset(mediaType) { (p, ps) ⇒ HttpEntity(mediaType withCharset ps, ByteString(p.toString)) }
  def personMarshaller(mediaType: MediaType.WithFixedCharset): ToEntityMarshaller[Person] =
    Marshaller.withFixedContentType(mediaType) { p ⇒ HttpEntity(mediaType, ByteString(p.toString)) }



  val john = Person(12,"John")
  val futP = Marshal(john).to[MessageEntity]

  val ps = (1 to 5).map {i => Person(i,s"member#i")}
  val source = Source(ps)

  import akka.http.scaladsl.common.{ EntityStreamingSupport, JsonEntityStreamingSupport }
  import akka.http.scaladsl.server.Directives._
  implicit val jsonStreamingSupport: JsonEntityStreamingSupport = EntityStreamingSupport.json()


  val route =
    path("data") {
 //     val fut = Marshal(source).to[Source[Person,NotUsed]]  //compile failed: implicit not found
      val fut2 = Marshal(source).toResponseFor(HttpRequest(method = HttpMethods.GET)) // compile ok!
      complete(source)  //ok due to magnet-patern type ToResponseMarshallable
    }

}

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

相关文章

网站地图xml地图