//设置package包名称以及导入依赖的类
package com.shashank.akkahttp.basic.routing

import akka.actor.ActorSystem
import akka.http.scaladsl.marshalling.Marshal
import akka.http.scaladsl.model.{HttpMethods, HttpRequest, HttpResponse, MessageEntity}
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.stream.{ActorMaterializer, Materializer}
import akka.util.ByteString

import scala.concurrent.Await
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global
import spray.json._


object UnMarshalling {

  def main(args: Array[String]) {

    implicit val sys = ActorSystem("IntroductionToAkkaHttp")
    implicit val mat:Materializer = ActorMaterializer()

    //type FromStringUnmarshaller[T] = Unmarshaller[String, T]
    val intFuture = Unmarshal("42").to[Int]
    val int = Await.result(intFuture, 1.second)
    println("int unmarshalling "+int)

    //type FromStringUnmarshaller[T] = Unmarshaller[String, T]
    val boolFuture = Unmarshal("off").to[Boolean]
    val bool = Await.result(boolFuture, 1.second)
    println("off unmarshalling "+bool)

    //type ToEntityMarshaller[T] = Marshaller[T, MessageEntity]
    val string = "Yeah"
    val entityFuture = Marshal(string).to[MessageEntity]
    val entity = Await.result(entityFuture, 1.second) // don't block in non-test code!
    println(entity)

    //type ToResponseMarshaller[T] = Marshaller[T, HttpResponse]
    val errorMsg = "Not found, pal!"
    val responseFuture = Marshal(404 -> errorMsg).to[HttpResponse]
    val response = Await.result(responseFuture, 1.second)
    println(response)


    //type FromEntityUnmarshaller[T] = Unmarshaller[HttpEntity, T]
    val jsonByteString = ByteString("""{"name":"Hello"}""")
    val httpRequest = HttpRequest(HttpMethods.POST, entity = jsonByteString)
    val jsonDataUnmarshalledFuture = Unmarshal(httpRequest).to[String]
    val jsonDataUnmarshalled = Await.result(jsonDataUnmarshalledFuture, 1.second)
    println(jsonDataUnmarshalled)

    sys.terminate()

  }

} 

//设置package包名称以及导入依赖的类
package com.ulasakdeniz.hakker

import akka.http.scaladsl.marshalling.{Marshal, ToEntityMarshaller}
import akka.http.scaladsl.model.{HttpHeader, HttpResponse, ResponseEntity, StatusCode}
import akka.http.scaladsl.server.Directives._
import akka.http.scaladsl.server.Route
import com.typesafe.config.ConfigFactory
import com.ulasakdeniz.hakker.template.Render
import de.heikoseeberger.akkahttpcirce.CirceSupport._
import io.circe.{Encoder, Json}
import io.circe.syntax._

import scala.collection.immutable
import scala.concurrent.ExecutionContext

trait Controller extends Render {

  override lazy val config = ConfigFactory.load()
  val StatusCodes          = akka.http.scaladsl.model.StatusCodes

  def route: Route

  def apply(): Route = {
    get {
      // render frontend files
      pathPrefix("js") {
        renderDir("js")
      }
    } ~ route
  }

  def send(statusCode: StatusCode): Route = complete(statusCode)

  def send[T](statusCode: StatusCode, content: T, headers: immutable.Seq[HttpHeader] = Nil)(
      implicit marshaller: ToEntityMarshaller[T],
      ec: ExecutionContext): Route = {
    val response = Marshal(content)
      .to[ResponseEntity](marshaller, ec)
      .map(entity => {
        HttpResponse(statusCode, headers = headers).withEntity(entity)
      })
    complete(response)
  }

  def sendJson[T](statusCode: StatusCode, content: T)(implicit encoder: Encoder[T],
                                                      ec: ExecutionContext): Route = {
    sendJson(statusCode, content.asJson)
  }

  def sendJson[T](content: T)(implicit encoder: Encoder[T], ec: ExecutionContext): Route = {
    sendJson(StatusCodes.OK, content)
  }

  def sendJson(statusCode: StatusCode, json: Json)(implicit ec: ExecutionContext): Route = {
    send(statusCode, Option(json.noSpaces))
  }
} 

//设置package包名称以及导入依赖的类
package net.ruippeixotog.scalafbp.protocol.registry

import scala.concurrent.{ ExecutionContext, Future }

import akka.actor.ActorSystem
import akka.event.slf4j.SLF4JLogging
import akka.http.scaladsl.Http
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
import akka.http.scaladsl.marshalling.Marshal
import akka.http.scaladsl.model.HttpMethods._
import akka.http.scaladsl.model.headers.{ Authorization, OAuth2BearerToken }
import akka.http.scaladsl.model.{ HttpRequest, HttpResponse, RequestEntity }
import akka.stream.Materializer
import fommil.sjs.FamilyFormats._

class RegistryClient(baseUrl: String = "http://api.flowhub.io") extends SLF4JLogging {

  def register(runtime: Runtime, token: String)(implicit system: ActorSystem, mat: Materializer, ec: ExecutionContext): Future[HttpResponse] = {
    log.debug(s"PUT $baseUrl/runtimes/${runtime.id}")
    Marshal(runtime).to[RequestEntity].flatMap { entity =>
      Http().singleRequest(HttpRequest(
        PUT,
        s"$baseUrl/runtimes/${runtime.id}",
        List(Authorization(OAuth2BearerToken(token))),
        entity))
    }
  }

  def unregister(runtimeId: String, token: String)(implicit system: ActorSystem, mat: Materializer): Future[HttpResponse] = {
    log.debug(s"DELETE $baseUrl/runtimes/$runtimeId")
    Http().singleRequest(HttpRequest(
      DELETE,
      s"$baseUrl/runtimes/$runtimeId",
      List(Authorization(OAuth2BearerToken(token)))))
  }
} 

//设置package包名称以及导入依赖的类
package com.tpalanga.test.account.api.users

import akka.actor.ActorSystem
import akka.http.scaladsl.marshalling.Marshal
import akka.http.scaladsl.model.{ContentTypes, RequestEntity}
import akka.stream.{ActorMaterializer, ActorMaterializerSettings, Materializer}
import com.tpalanga.test.account.api.users.model.{NewUser, User, Users}
import com.tpalanga.test.config.TestConfig
import com.tpalanga.testlib.test.client.{NoEntity, Response, RestServiceClient}
import com.tpalanga.testlib.test.config.RestServiceConfig
import com.typesafe.scalalogging.LazyLogging

import scala.concurrent.{ExecutionContext, Future}

class AccountServiceRestClient(val restServiceConfig: RestServiceConfig)
                              (implicit val testConfig: TestConfig, val system: ActorSystem)
extends RestServiceClient with LazyLogging {
  import NoEntity.DataFormats._
  import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
  import com.tpalanga.test.account.api.users.model.UserJsonProtocol._

  logger.debug(s"AccountServiceRestServiceClient: $restServiceConfig")
  private implicit val materializer: Materializer = ActorMaterializer(ActorMaterializerSettings(system))

  def userRetrieve(id: String)(implicit ec: ExecutionContext): Future[Response[User]] =
    client.get(s"/data/users/$id").map { httpResponse =>
      Response[User](httpResponse)
    }

  def userCreate(user: NewUser)(implicit ec: ExecutionContext): Future[Response[User]] =
    for {
      entity <- Marshal(user).to[RequestEntity]
      httpResponse <- client.post(s"/data/users", Nil, entity.withContentType(ContentTypes.`application/json`))
    } yield Response[User](httpResponse)

  def userUpdate(user: User)(implicit ec: ExecutionContext): Future[Response[User]] =
    for {
      entity <- Marshal(user).to[RequestEntity]
      httpResponse <- client.put(s"/data/users/${user.id}", Nil, entity.withContentType(ContentTypes.`application/json`))
    } yield Response[User](httpResponse)

  def userDelete(id: String)(implicit ec: ExecutionContext): Future[Response[NoEntity]] =
    client.delete(s"/data/users/$id").map { httpResponse =>
      Response[NoEntity](httpResponse)
    }

  def userList()(implicit ec: ExecutionContext): Future[Response[Users]] =
    client.get(s"/data/users").map { httpResponse =>
      Response[Users](httpResponse)
    }
} 

//设置package包名称以及导入依赖的类
package com.tpalanga.testlib.test.client.impl

import akka.actor.ActorSystem
import akka.http.scaladsl.marshalling.Marshal
import akka.http.scaladsl.model.{ContentTypes, RequestEntity}
import akka.stream.{ActorMaterializer, ActorMaterializerSettings, Materializer}
import com.tpalanga.testlib.test.client.{NoEntity, Response, RestServiceClient}
import com.tpalanga.testlib.test.config.RestServiceConfig
import com.typesafe.scalalogging.LazyLogging

import scala.concurrent.{ExecutionContext, Future}


object NewsletterServiceRestClient {
  type NewsletterServiceRestClientFactory = (RestServiceConfig, ActorSystem) => NewsletterServiceRestClient

  def defaultFactory: NewsletterServiceRestClientFactory =
    (config, system) => new NewsletterServiceRestClient(config)(system)

}

class NewsletterServiceRestClient(val restServiceConfig: RestServiceConfig)
                                 (implicit val system: ActorSystem)
  extends RestServiceClient with LazyLogging {
  import NoEntity.DataFormats._
  import SubscriberJsonProtocol._
  import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._

  logger.debug(s"NewsletterServiceRestServiceClient: $restServiceConfig")
  private implicit val materializer: Materializer = ActorMaterializer(ActorMaterializerSettings(system))

  def subscriberRetrieve(id: String)(implicit ec: ExecutionContext): Future[Response[Subscriber]] =
    client.get(s"/data/subscribers/$id").map { httpResponse =>
      Response[Subscriber](httpResponse)
    }

  def subscriberCreate(subscriber: Subscriber)(implicit ec: ExecutionContext): Future[Response[Subscriber]] =
    for {
      entity <- Marshal(subscriber).to[RequestEntity]
      httpResponse <- client.post(s"/data/subscribers", Nil, entity.withContentType(ContentTypes.`application/json`))
    } yield Response[Subscriber](httpResponse)

  def subscriberUpdate(user: Subscriber)(implicit ec: ExecutionContext): Future[Response[Subscriber]] =
    for {
      entity <- Marshal(user).to[RequestEntity]
      httpResponse <- client.put(s"/data/subscribers/${user.id}", Nil, entity.withContentType(ContentTypes.`application/json`))
    } yield Response[Subscriber](httpResponse)

  def subscriberDelete(id: String)(implicit ec: ExecutionContext): Future[Response[NoEntity]] =
    client.delete(s"/data/subscribers/$id").map { httpResponse =>
      Response[NoEntity](httpResponse)
    }

  def subscriberList()(implicit ec: ExecutionContext): Future[Response[Subscribers]] =
    client.get(s"/data/subscribers").map { httpResponse =>
      Response[Subscribers](httpResponse)
    }

} 

//设置package包名称以及导入依赖的类
package com.tpalanga.test.newsletter.api.subscriber

import akka.http.scaladsl.marshalling.Marshal
import akka.http.scaladsl.model.{ContentTypes, RequestEntity}
import akka.stream.{ActorMaterializer, ActorMaterializerSettings, Materializer}
import com.tpalanga.test.config.TestConfig
import com.tpalanga.test.newsletter.api.subscriber.model.{Subscriber, Subscribers}
import com.tpalanga.testlib.test.client.{NoEntity, Response, RestServiceClient}
import com.tpalanga.testlib.test.config.RestServiceConfig

import scala.concurrent.{ExecutionContext, Future}

trait NewsletterServiceRestServiceClient extends RestServiceClient {
  import NoEntity.DataFormats._
  import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
  import com.tpalanga.test.newsletter.api.subscriber.model.SubscriberJsonProtocol._

  val testConfig: TestConfig

  override val restServiceConfig: RestServiceConfig = testConfig.restServiceConfig
  private implicit val materializer: Materializer = ActorMaterializer(ActorMaterializerSettings(system))

  def subscriberRetrieve(id: String)(implicit ec: ExecutionContext): Future[Response[Subscriber]] =
    client.get(s"/data/subscribers/$id").map { httpResponse =>
      Response[Subscriber](httpResponse)
    }

  def subscriberCreate(subscriber: Subscriber)(implicit ec: ExecutionContext): Future[Response[Subscriber]] =
    for {
      entity <- Marshal(subscriber).to[RequestEntity]
      httpResponse <- client.post(s"/data/subscribers", Nil, entity.withContentType(ContentTypes.`application/json`))
    } yield Response[Subscriber](httpResponse)

  def subscriberUpdate(user: Subscriber)(implicit ec: ExecutionContext): Future[Response[Subscriber]] =
    for {
      entity <- Marshal(user).to[RequestEntity]
      httpResponse <- client.put(s"/data/subscribers/${user.id}", Nil, entity.withContentType(ContentTypes.`application/json`))
    } yield Response[Subscriber](httpResponse)

  def subscriberDelete(id: String)(implicit ec: ExecutionContext): Future[Response[NoEntity]] =
    client.delete(s"/data/subscribers/$id").map { httpResponse =>
      Response[NoEntity](httpResponse)
    }

  def subscriberList()(implicit ec: ExecutionContext): Future[Response[Subscribers]] =
    client.get(s"/data/subscribers").map { httpResponse =>
      Response[Subscribers](httpResponse)
    }

} 

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

import java.io.File

import akka.NotUsed
import akka.actor.ActorSystem
import akka.http.scaladsl.marshalling.Marshal
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{FileIO, Source}
import akka.util.ByteString

import scaladays.akka.domain.Tweet
import scaladays.akka.http.MyJsonProtocol
import scaladays.akka.support.MakingUpData

object Step0GenerateStatuses extends App
  with MakingUpData with MyJsonProtocol {

  implicit val system = ActorSystem()
  implicit val ec = system.dispatcher
  implicit val mat = ActorMaterializer()

  val n = 10000

  Source.repeat(NotUsed).take(n)
    .map(_ => Tweet.random)
    .mapAsync(1)(t => Marshal(t).to[ByteString])
    .intersperse(ByteString("\n"))
    .runWith(FileIO.toPath(new File("tweets.json").toPath))
    .onComplete { res =>
      println(s"Generated $n tweets. ($res)")
      system.terminate()
    }

} 

//设置package包名称以及导入依赖的类
package reactive.queue.router

import akka.actor.{ActorLogging, Props}
import akka.http.scaladsl.Http
import akka.http.scaladsl.marshalling.Marshal
import akka.http.scaladsl.model._
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.pattern.pipe
import akka.stream.ActorMaterializer
import akka.stream.actor.ActorSubscriberMessage.{OnComplete, OnError, OnNext}
import akka.stream.actor.{ActorSubscriber, OneByOneRequestStrategy, RequestStrategy}
import akka.util.ByteString
import io.scalac.amqp.{Connection, Message}

object QueueSubscriber {
  def props(queueConnection: Connection, queueName: String, queueSubscriberUri: Uri): Props =
    Props(classOf[QueueSubscriber],queueConnection,  queueName, queueSubscriberUri)
}

class QueueSubscriber(queueConnection: Connection, queueName: String, queueSubscriberUri: Uri) extends
  ActorSubscriber with ActorLogging {
  implicit val system = context.system
  implicit val ec = context.dispatcher
  implicit val materlizer = ActorMaterializer()

  override protected def requestStrategy: RequestStrategy = OneByOneRequestStrategy

  def receive = {
    case OnNext(message: Message) => route(ByteString(message.body.toArray).decodeString("UTF-8"))
    case OnComplete => log.info("*** on complete")
    case OnError(error) => log.error(s"*** on error: $error")
    case HttpResponse(status, _, _, _) => log.info(s"*** route response: ${status.intValue}")
  }

  def route(message: String): Unit = {
    log.info(s"*** on next: $message")
    try {
      val httpResponse = for {
        request <- Marshal(message).to[RequestEntity]
        response <- Http().singleRequest(HttpRequest(method = HttpMethods.POST, uri = queueSubscriberUri, entity = request))
        entity <- Unmarshal(response).to[HttpResponse]
      } yield entity
      httpResponse.pipeTo(self)
    } catch {
      case t: Throwable =>
        log.error(s"*** on next: forward to uri $queueSubscriberUri failed on: $message with error: ${t.getMessage}")
        queueConnection.publish(queueName, message)
        log.info(s"*** on next: republished to queue $queueName")
    }
  }
} 

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

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.marshalling.Marshal
import akka.http.scaladsl.model._
import akka.http.scaladsl.model.headers.{ Accept, Authorization, OAuth2BearerToken }
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.stream.Materializer
import com.jatescher.layer.http.MediaRanges.LayerJsonMediaRange
import com.jatescher.layer.marshalling.Marshallers.{ ErrorResponseUnmarshaller, _ }
import com.jatescher.layer.models.{ ErrorResponse, Message }
import com.typesafe.config.Config
import scala.concurrent.{ ExecutionContext, Future }

class LayerClient(router: LayerRouter, config: Config)(implicit system: ActorSystem, materializer: Materializer, ec: ExecutionContext) {
  val LAYER_TOKEN = config.getString("layer.token")

  def sendMessage(message: Message): Future[Either[ErrorResponse, Message]] = {
    for {
      messageRequest <- sendMessageRequest(message)
      response <- executeRequest(messageRequest)
      messageOrErrorResponse <- unmarshallResponse(response)
    } yield messageOrErrorResponse
  }

  private def sendMessageRequest(message: Message): Future[HttpRequest] = {
    Marshal(message).to[RequestEntity].map { entity =>
      HttpRequest(
        method = HttpMethods.POST,
        uri = router.createMessageUrl(message.conversation),
        entity = entity,
        headers = List(
          Authorization(OAuth2BearerToken(LAYER_TOKEN)),
          Accept(LayerJsonMediaRange)
        )
      )
    }
  }

  protected def executeRequest(httpRequest: HttpRequest): Future[HttpResponse] = {
    Http().singleRequest(httpRequest)
  }

  private def unmarshallResponse(response: HttpResponse): Future[Either[ErrorResponse, Message]] = {
    val unmarshalledResponse = Unmarshal(response.entity)

    if (response.status == StatusCodes.Created) {
      unmarshalledResponse.to[Message].map(Right(_))
    } else {
      unmarshalledResponse.to[ErrorResponse].map(Left(_))
    }
  }

} 

//设置package包名称以及导入依赖的类
package com.bluelabs.s3stream


import scala.concurrent.{ExecutionContext, Future}
import akka.http.scaladsl.marshallers.xml.ScalaXmlSupport._
import akka.http.scaladsl.marshalling.Marshal
import akka.http.scaladsl.model._
import akka.http.scaladsl.model.Uri.Query
import akka.http.scaladsl.model.headers.Host
import akka.util.ByteString

object HttpRequests {

  def s3Request(s3Location: S3Location, method: HttpMethod = HttpMethods.GET, uriFn: (Uri => Uri) = identity): HttpRequest = {
    HttpRequest(method)
      .withHeaders(Host(requestHost(s3Location)))
      .withUri(uriFn(requestUri(s3Location)))
  }

  def initiateMultipartUploadRequest(s3Location: S3Location): HttpRequest = {
    s3Request(s3Location, HttpMethods.POST, _.withQuery(Query("uploads")))
  }
  
  def getRequest(s3Location: S3Location): HttpRequest = {
    s3Request(s3Location)
  }

  def uploadPartRequest(upload: MultipartUpload, partNumber: Int, payload: ByteString): HttpRequest = {
    s3Request(upload.s3Location,
              HttpMethods.PUT,
              _.withQuery(Query("partNumber" -> partNumber.toString, "uploadId" -> upload.uploadId))
    ).withEntity(payload)
  }

  def completeMultipartUploadRequest(upload: MultipartUpload, parts: Seq[(Int, String)])(implicit ec: ExecutionContext): Future[HttpRequest] = {
    val payload = <CompleteMultipartUpload>
        {
          parts.map{case (partNumber, etag) => <Part><PartNumber>{partNumber}</PartNumber><ETag>{etag}</ETag></Part>}
        }
      </CompleteMultipartUpload>
    for {
      entity <- Marshal(payload).to[RequestEntity]
    } yield {
      s3Request(upload.s3Location,
                HttpMethods.POST,
                _.withQuery(Query("uploadId" -> upload.uploadId))
      ).withEntity(entity)
    }
  }


  def requestHost(s3Location: S3Location): Uri.Host = Uri.Host(s"${s3Location.bucket}.s3.amazonaws.com")

  def requestUri(s3Location: S3Location): Uri = Uri(s"/${s3Location.key}").withHost(requestHost(s3Location)).withScheme("https")
} 

//设置package包名称以及导入依赖的类
package mm4s.api

import akka.NotUsed
import akka.actor.ActorSystem
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
import akka.http.scaladsl.marshalling.Marshal
import akka.http.scaladsl.model._
import akka.stream.scaladsl.Source
import spray.json.{DefaultJsonProtocol, RootJsonFormat}


object Teams {
  import Streams._
  import TeamModels._
  import TeamProtocols._

  def create(team: CreateTeam)(implicit system: ActorSystem): Source[HttpRequest, NotUsed] = {
    request("/teams/create") { r =>
      Marshal(team).to[MessageEntity].map(r.withMethod(HttpMethods.POST).withEntity)
    }
  }

  def find(name: String, token: String)(implicit system: ActorSystem): Source[HttpRequest, NotUsed] = {
    get("/teams/find_team_by_name").map(r =>
      r.withMethod(HttpMethods.POST).withHeaders(auth(token)).withEntity(s"""{"name":"$name"}""")
    )
  }

  def list(token: String)(implicit system: ActorSystem): Source[HttpRequest, NotUsed] = {
    get("/teams/all").map(withAuth(token))
  }
}

object TeamModels {
  case class Team(id: String, name: String)
  case class CreateTeam(display_name: String, name: String, email: String, `type`: String = "O")
  case class TeamCreated(id: String, display_name: String, name: String, email: String)
}

object TeamProtocols extends DefaultJsonProtocol with SprayJsonSupport {
  import TeamModels._

  implicit val CreateTeamFormat: RootJsonFormat[CreateTeam] = jsonFormat4(CreateTeam)
  implicit val TeamCreatedFormat: RootJsonFormat[TeamCreated] = jsonFormat4(TeamCreated)
  implicit val TeamFormat: RootJsonFormat[Team] = jsonFormat2(Team)
} 

//设置package包名称以及导入依赖的类
package mm4s.api

import java.nio.file.Path

import akka.NotUsed
import akka.actor.ActorSystem
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
import akka.http.scaladsl.marshalling.Marshal
import akka.http.scaladsl.model._
import akka.stream.scaladsl.Source
import spray.json.{DefaultJsonProtocol, RootJsonFormat}



object Messages {
  import MessageModels._
  import MessageProtocols._
  import Streams._

  def create(post: CreatePost, token: String)(implicit system: ActorSystem) = {
    request(s"/channels/${post.channel_id}/create") { r =>
      Marshal(post).to[MessageEntity].map(r.withMethod(HttpMethods.POST).withHeaders(auth(token)).withEntity)
    }
  }

  def since(time: Long, channel_id: String, token: String)(implicit system: ActorSystem): Source[HttpRequest, NotUsed] = {
    get(s"/channels/$channel_id/posts/$time").map(withAuth(token))
  }
}

object MessageModels {
  case class CreatePost(message: String, channel_id: String, filenames: Seq[String] = Seq.empty)
  case class Posting(user_id: String, channel_id: String, message: String, hashtags: String, create_at: Long, filenames: Seq[String])
  case class PostQueryResponse(order: Seq[String], posts: Option[Map[String, Posting]])
}

object MessageProtocols extends DefaultJsonProtocol with SprayJsonSupport {
  import MessageModels._

  implicit val CreatePostFormat: RootJsonFormat[CreatePost] = jsonFormat3(CreatePost)
  implicit val PostingFormat: RootJsonFormat[Posting] = jsonFormat6(Posting)
  implicit val PostQueryResponseFormat: RootJsonFormat[PostQueryResponse] = jsonFormat2(PostQueryResponse)
} 

//设置package包名称以及导入依赖的类
package mm4s.api

import akka.actor.ActorSystem
import akka.http.scaladsl.marshalling.Marshal
import akka.http.scaladsl.model.HttpMethods._
import akka.http.scaladsl.model.{HttpRequest, RequestEntity}
import akka.stream.ActorMaterializer
import akka.stream.testkit.scaladsl.TestSink
import akka.testkit.TestKit
import mm4s.api.UserModels.{CreateUser, LoginByUsername}
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.{Matchers, WordSpecLike}


class UsersSpec extends TestKit(ActorSystem("UsersSpec"))
                        with WordSpecLike with Matchers with ScalaFutures {

  implicit val mat = ActorMaterializer()

  "api calls" should {
    "have proper paths" when {
      "create" in {
        import UserProtocols._
        val o = CreateUser("", "", "", "")
        val e = Marshal(o).to[RequestEntity].futureValue

        val path = uripath("/users/create")

        Users.create(o)
        .runWith(TestSink.probe[HttpRequest])
        .request(1)
        .expectNext(HttpRequest(uri = path, method = POST, entity = e))
        .expectComplete()
      }

      "login" in {
        import UserProtocols._
        val o = LoginByUsername("", "", "")
        val e = Marshal(o).to[RequestEntity].futureValue

        val path = uripath("/users/login")

        Users.login(o)
        .runWith(TestSink.probe[HttpRequest])
        .request(1)
        .expectNext(HttpRequest(uri = path, method = POST, entity = e))
        .expectComplete()
      }
    }
  }
} 

//设置package包名称以及导入依赖的类
package mm4s.api

import akka.actor.ActorSystem
import akka.http.scaladsl.marshalling.Marshal
import akka.http.scaladsl.model.HttpMethods._
import akka.http.scaladsl.model.headers.Cookie
import akka.http.scaladsl.model.{HttpMethods, HttpRequest, RequestEntity}
import akka.stream.ActorMaterializer
import akka.stream.testkit.scaladsl.TestSink
import akka.testkit.TestKit
import mm4s.api.TeamModels.CreateTeam
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.{Matchers, WordSpecLike}


class TeamsSpec extends TestKit(ActorSystem("TeamsSpec"))
                        with WordSpecLike with Matchers with ScalaFutures {

  implicit val mat = ActorMaterializer()

  "api calls" should {
    "have proper paths" when {
      "create" in {
        import TeamProtocols._
        val o = CreateTeam("", "", "")
        val e = Marshal(o).to[RequestEntity].futureValue

        val path = uripath("/teams/create")

        Teams.create(o)
        .runWith(TestSink.probe[HttpRequest])
        .request(1)
        .expectNext(HttpRequest(uri = path, method = POST, entity = e))
        .expectComplete()
      }
      "all" in {
        val path = uripath("/teams/all")

        Teams.list("token")
        .runWith(TestSink.probe[HttpRequest])
        .request(1)
        .expectNext(HttpRequest(uri = path, method = GET, headers = List(auth("token"))))
        .expectComplete()
      }
    }
  }
} 

//设置package包名称以及导入依赖的类
package mm4s

import akka.actor.ActorSystem
import akka.http.scaladsl.marshalling.Marshal
import akka.http.scaladsl.model.MessageEntity
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Sink
import mm4s.api.{UserProtocols, UserModels}
import UserModels.CreateUser
import org.scalactic.FutureSugar
import org.scalatest._
import spray.json._


class MarshallingSpec extends AsyncWordSpec with Matchers with BeforeAndAfterAll with FutureSugar {
  implicit val system = ActorSystem()
  implicit val materializer = ActorMaterializer()
  import system.dispatcher


  "user models" should {
    import UserProtocols._

    val createUserJson =
      """{
        |"username":"bob",
        |"password":"pass",
        |"email":"[email protected]",
        |"team_id":"team-id"
        |}""".stripMargin

    "CreateUser" in {
      val m = CreateUser("bob", "pass", "[email protected]", "team-id")
      Marshal(m).to[MessageEntity].flatMap(_.dataBytes.map(_.utf8String).runWith(Sink.head)).map { r =>
        r.parseJson shouldBe createUserJson.parseJson
      }
    }
  }

  override protected def afterAll(): Unit = system.terminate()
} 

//设置package包名称以及导入依赖的类
package fr.acinq.eclair.blockchain.rpc

import java.io.IOException

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.marshalling.Marshal
import akka.http.scaladsl.model._
import akka.http.scaladsl.model.headers.{Authorization, BasicHttpCredentials}
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.stream.ActorMaterializer
import de.heikoseeberger.akkahttpjson4s.Json4sSupport._
import org.json4s.JsonAST.JValue
import org.json4s.{DefaultFormats, jackson}

import scala.concurrent.{ExecutionContext, Future}

// @formatter:off
case class JsonRPCRequest(jsonrpc: String = "1.0", id: String = "scala-client", method: String, params: Seq[Any])
case class Error(code: Int, message: String)
case class JsonRPCResponse(result: JValue, error: Option[Error], id: String)
case class JsonRPCError(error: Error) extends IOException(s"${error.message} (code: ${error.code})")
// @formatter:on

class BitcoinJsonRPCClient(user: String, password: String, host: String = "127.0.0.1", port: Int = 8332, ssl: Boolean = false)(implicit system: ActorSystem) {

  val scheme = if (ssl) "https" else "http"
  val uri = Uri(s"$scheme://$host:$port")

  implicit val materializer = ActorMaterializer()
  val httpClient = Http(system)
  implicit val serialization = jackson.Serialization
  implicit val formats = DefaultFormats

  def invoke(method: String, params: Any*)(implicit ec: ExecutionContext): Future[JValue] =
    for {
      entity <- Marshal(JsonRPCRequest(method = method, params = params)).to[RequestEntity]
      httpRes <- httpClient.singleRequest(HttpRequest(uri = uri, method = HttpMethods.POST).addHeader(Authorization(BasicHttpCredentials(user, password))).withEntity(entity))
      jsonRpcRes <- Unmarshal(httpRes).to[JsonRPCResponse].map {
        case JsonRPCResponse(_, Some(error), _) => throw JsonRPCError(error)
        case o => o
      } recover {
        case t: Throwable if httpRes.status == StatusCodes.Unauthorized => throw new RuntimeException("bitcoind replied with 401/Unauthorized (bad user/password?)", t)
      }
    } yield jsonRpcRes.result

  def invoke(request: Seq[(String, Seq[Any])])(implicit ec: ExecutionContext): Future[Seq[JValue]] =
    for {
      entity <- Marshal(request.map(r => JsonRPCRequest(method = r._1, params = r._2))).to[RequestEntity]
      httpRes <- httpClient.singleRequest(HttpRequest(uri = uri, method = HttpMethods.POST).addHeader(Authorization(BasicHttpCredentials(user, password))).withEntity(entity))
      jsonRpcRes <- Unmarshal(httpRes).to[Seq[JsonRPCResponse]].map {
        //case JsonRPCResponse(_, Some(error), _) => throw JsonRPCError(error)
        case o => o
      } recover {
        case t: Throwable if httpRes.status == StatusCodes.Unauthorized => throw new RuntimeException("bitcoind replied with 401/Unauthorized (bad user/password?)", t)
      }
    } yield jsonRpcRes.map(_.result)

} 

//设置package包名称以及导入依赖的类
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
import akka.http.scaladsl.marshalling.Marshal
import akka.http.scaladsl.model.headers.{Authorization, BasicHttpCredentials}
import akka.http.scaladsl.model.{HttpMethods, HttpRequest, RequestEntity}
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.stream.ActorMaterializer
import de.choffmeister.auth.common.OAuth2AccessTokenResponse
import model.{CreateUserRequest, JsonProtocols, UserInfo}

import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global

package object UserTestUtils extends JsonProtocols {

  implicit val system       = ActorSystem()
  implicit val materializer = ActorMaterializer()

  def createUser(createUserRequest: CreateUserRequest): Future[UserInfo] = {
    for {
      request <- Marshal(createUserRequest).to[RequestEntity]
      response <- Http().singleRequest(
        HttpRequest(method = HttpMethods.POST, uri = s"http://localhost:9000/user/create", entity = request))
      userInfo <- Unmarshal(response.entity).to[UserInfo]
    } yield userInfo
  }

  def authenticateUser(userInfo: UserInfo): Future[OAuth2AccessTokenResponse] = {
    for {
      response <- Http().singleRequest(
        HttpRequest(uri = s"http://localhost:9000/token/create",
                    headers = List(Authorization(BasicHttpCredentials(userInfo.userName, "password")))))
      accessToken <- Unmarshal(response.entity).to[OAuth2AccessTokenResponse]
    } yield accessToken
  }

} 

//设置package包名称以及导入依赖的类
package akkaviz.server

import akka.http.scaladsl.marshalling.{Marshal, Marshaller}
import akka.http.scaladsl.model.HttpEntity.ChunkStreamPart
import akka.http.scaladsl.model.{HttpEntity, HttpResponse, MediaTypes}
import akka.http.scaladsl.server.{Directives, StandardRoute}
import akka.stream.scaladsl.{Flow, Source}

import scala.concurrent.ExecutionContext

trait AkkaHttpHelpers {

  def asJsonArray[T](implicit m: Marshaller[T, String], ec: ExecutionContext): Flow[T, HttpEntity.ChunkStreamPart, _] = {
    Flow.apply[T]
      .mapAsync[String](4)(t => Marshal(t).to[String])
      .scan[Option[ChunkStreamPart]](None) {
        case (None, s: String) => Some(ChunkStreamPart(s))
        case (_, s: String)    => Some(ChunkStreamPart(s",${s}"))
      }.mapConcat(_.toList)
      .prepend(Source.single(ChunkStreamPart("[")))
      .concat(Source.single(ChunkStreamPart("]")))
  }

  def completeAsJson[T](source: Source[T, _])(implicit m: Marshaller[T, String], ec: ExecutionContext): StandardRoute = {
    Directives.complete(HttpResponse(
      entity = HttpEntity.Chunked(MediaTypes.`application/json`, source.via(asJsonArray))
    ))
  }
}

object AkkaHttpHelpers extends AkkaHttpHelpers 

//设置package包名称以及导入依赖的类
package im.actor.server.sms

import scala.concurrent.{ ExecutionContext, Future }

import akka.actor.ActorSystem
import akka.http.scaladsl.HttpExt
import akka.http.scaladsl.marshalling.Marshal
import akka.http.scaladsl.model._
import akka.http.scaladsl.model.headers.{ Authorization, BasicHttpCredentials }
import akka.stream.Materializer
import com.typesafe.config.Config

class TwilioSmsEngine(config: Config)(implicit system: ActorSystem, materializer: Materializer, http: HttpExt) extends SmsEngine {

  private val account = config.getString("account")
  private val token = config.getString("token")
  private val from = config.getString("from")

  implicit val ec: ExecutionContext = system.dispatcher

  private val baseUri = Uri(s"https://api.twilio.com/2010-04-01/Accounts/$account/Messages.json")

  private val authHeader = Authorization(BasicHttpCredentials(account, token))

  override def send(phoneNumber: Long, message: String): Future[Unit] = {
    val to = s"+${phoneNumber}"

    Marshal(FormData(Map("From" ? from, "To" ? to, "Body" ? message))).to[RequestEntity] flatMap { entity ?
      val request = HttpRequest(
        method = HttpMethods.POST,
        uri = baseUri,
        entity = entity
      ).withHeaders(authHeader)

      http.outgoingConnection("api.twilio.com", 443)

      val f = http.singleRequest(request) map {
        case HttpResponse(StatusCodes.Created, _, _, _) ? ()
        case resp ?
          throw new Exception(s"Wrong response: ${resp}")
      }

      f onFailure {
        case e ?
          system.log.error(e, "Failed to send sms through twilio")
      }

      f
    }
  }
}