//设置package包名称以及导入依赖的类
package akka.stream.alpakka.dynamodb.impl

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.MediaType.NotCompressible
import akka.http.scaladsl.model.{ContentType, MediaType}
import akka.stream.Materializer
import akka.stream.alpakka.dynamodb.AwsOp
import akka.stream.alpakka.dynamodb.impl.AwsClient.{AwsConnect, AwsRequestMetadata}
import akka.stream.scaladsl.{Sink, Source}
import com.amazonaws.AmazonServiceException
import com.amazonaws.http.HttpResponseHandler

class DynamoClientImpl(
    val settings: DynamoSettings,
    val errorResponseHandler: HttpResponseHandler[AmazonServiceException]
)(implicit protected val system: ActorSystem, implicit protected val materializer: Materializer)
    extends AwsClient[DynamoSettings] {

  override protected val service = "dynamodb"
  override protected val defaultContentType =
    ContentType.Binary(MediaType.customBinary("application", "x-amz-json-1.0", NotCompressible))
  override protected implicit val ec = system.dispatcher

  override protected val connection: AwsConnect =
    if (settings.port == 443)
      Http().cachedHostConnectionPoolHttps[AwsRequestMetadata](settings.host)(materializer)
    else
      Http().cachedHostConnectionPool[AwsRequestMetadata](settings.host, settings.port)(materializer)

  def single(op: AwsOp) = Source.single(op).via(flow).map(_.asInstanceOf[op.B]).runWith(Sink.head)

} 

//设置package包名称以及导入依赖的类
package com.jatescher.layer.marshalling.v1

import akka.http.scaladsl.model.MediaType
import spray.json._

object MediaTypeFormat extends DefaultJsonProtocol {

  implicit val MediaTypeJsonFormat: JsonFormat[MediaType] = new JsonFormat[MediaType] {
    def write(mediaType: MediaType) = mediaType.toString.toJson
    def read(value: JsValue) = value match {
      case JsString(mediaTypeString) => MediaType.parse(mediaTypeString) match {
        case Right(mediaType) => mediaType
        case _ => deserializationError("Valid media type expected")
      }
      case _ => deserializationError("Media type cannot be blank")
    }
  }

} 

//设置package包名称以及导入依赖的类
package com.jatescher.layer.models

import akka.http.scaladsl.model.{ MediaType, MediaTypes }

sealed trait MessagePart {
  val mimeType: MediaType
}

object MessageParts {

  // Simple text message part
  case class TextMessagePart(body: String) extends MessagePart {
    val mimeType = MediaTypes.`text/plain`
  }

  // Custom content message part
  case class ImageMessagePart(content: MessagePartContent) extends MessagePart {
    val mimeType = content.mimeType
  }

} 

//设置package包名称以及导入依赖的类
package com.jatescher.layer.models

import akka.http.scaladsl.model.MediaType

sealed trait MessagePartContent {
  val id: String
  val mimeType: MediaType
}

object MessagePartContents {

  // Rich content message part
  case class ImageMessagePartContent(
    id: String,
    downloadUrl: String,
    expiration: String,
    refreshUrl: String,
    size: Long,
    mimeType: MediaType
  ) extends MessagePartContent

} 

//设置package包名称以及导入依赖的类
package com.jatescher.layer.marshalling

import akka.http.scaladsl.model.{ MediaType, MediaTypes }
import com.jatescher.layer.marshalling.v1.MediaTypeFormat._
import org.scalatest.{ Matchers, WordSpec }
import spray.json._

class MediaTypeFormatSpec extends WordSpec with Matchers {
  val mediaTypeString = "image/png"
  val mediaType: MediaType = MediaTypes.`image/png`

  "#write" should {
    "marshall the media type as string" in {
      mediaType.toJson shouldBe mediaTypeString.toJson
    }
  }

  "#read" should {
    "unmarshall media types" in {
      JsString(mediaTypeString).convertTo[MediaType] shouldBe mediaType
    }

    "raise an exception if the media type is not a string" in {
      intercept[DeserializationException] {
        JsNumber(1).convertTo[MediaType]
      }
    }

    "raise an exception if the media type is not valid" in {
      intercept[DeserializationException] {
        JsString("non-media-type").convertTo[MediaType]
      }
    }
  }

} 

//设置package包名称以及导入依赖的类
package com.example.akka.http

import java.io.{ OutputStreamWriter, PipedInputStream, PipedOutputStream }

import scala.concurrent.{ ExecutionContext, Future }

import akka.http.scaladsl.model.{ HttpCharsets, HttpEntity, MediaType }
import akka.http.scaladsl.server.{ Directives, Route }
import akka.stream.scaladsl.StreamConverters
import io.prometheus.client.CollectorRegistry
import io.prometheus.client.exporter.common.TextFormat

object PrometheusService extends Directives {

  lazy val prometheusTextType =
    MediaType.customWithFixedCharset("text", "plain", HttpCharsets.`UTF-8`, params = Map("version" -> "0.0.4"))

  def route(implicit executionContext: ExecutionContext): Route = {
    path("metrics") {
      complete {
        val in = new PipedInputStream
        val out = new OutputStreamWriter(new PipedOutputStream(in), HttpCharsets.`UTF-8`.value)
        val byteSource = StreamConverters.fromInputStream(() => in)
        Future {
          try {
            TextFormat.write004(out, CollectorRegistry.defaultRegistry.metricFamilySamples())
            out.flush()
          } finally {
            out.close()
          }
        }
        HttpEntity(prometheusTextType, byteSource)
      }
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.example.utilities.serialization

import akka.http.scaladsl.marshalling.{Marshaller, ToEntityMarshaller}
import akka.http.scaladsl.model.MediaType.Compressible
import akka.http.scaladsl.model.{ContentType, ContentTypes, HttpEntity, MediaType}
import akka.http.scaladsl.unmarshalling.Unmarshaller.UnsupportedContentTypeException
import akka.http.scaladsl.unmarshalling.{FromEntityUnmarshaller, Unmarshaller}
import akka.http.scaladsl.util.FastFuture
import com.google.protobuf.CodedInputStream
import com.trueaccord.scalapb.json.JsonFormat
import com.trueaccord.scalapb.{GeneratedMessage, GeneratedMessageCompanion, Message}

import scala.concurrent.Future

trait ScalaPBMarshalling {
  private val protobufContentType = ContentType(MediaType.applicationBinary("octet-stream", Compressible, "proto"))
  private val applicationJsonContentType = ContentTypes.`application/json`

  def scalaPBFromRequestUnmarshaller[O <: GeneratedMessage with Message[O]](companion: GeneratedMessageCompanion[O]): FromEntityUnmarshaller[O] = {
    Unmarshaller.withMaterializer[HttpEntity, O](_ => implicit mat => {
      case [email protected](`applicationJsonContentType`, data) =>
        val charBuffer = Unmarshaller.bestUnmarshallingCharsetFor(entity)
        FastFuture.successful(JsonFormat.fromJsonString(data.decodeString(charBuffer.nioCharset().name()))(companion))
      case [email protected](`protobufContentType`, data) =>
        FastFuture.successful(companion.parseFrom(CodedInputStream.newInstance(data.asByteBuffer)))
      case entity =>
        Future.failed(UnsupportedContentTypeException(applicationJsonContentType, protobufContentType))
    })
  }

  implicit def scalaPBToEntityMarshaller[U <: GeneratedMessage]: ToEntityMarshaller[U] = {
    def jsonMarshaller(): ToEntityMarshaller[U] = {
      val contentType = applicationJsonContentType
      Marshaller.withFixedContentType(contentType) { value =>
        HttpEntity(contentType, JsonFormat.toJsonString(value))
      }
    }

    def protobufMarshaller(): ToEntityMarshaller[U] = {
      Marshaller.withFixedContentType(protobufContentType) { value =>
        HttpEntity(protobufContentType, value.toByteArray)
      }
    }

    Marshaller.oneOf(jsonMarshaller(), protobufMarshaller())
  }

}