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

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.stream.ActorMaterializer
import com.typesafe.config.ConfigFactory
import scala.concurrent.{ExecutionContext, Future}

class StatisticDataFetcher(implicit ec: ExecutionContext, system: ActorSystem, materializer: ActorMaterializer) extends AutoMarshaller {

  val statisticsServiceUrl = {
    val config = ConfigFactory.load()
    config.getString("statisticsServiceUrl")
  }

  def getStatistics(): Future[List[StatisticData]] = {
    implicit val serialization = this.serialization
    implicit val formats = this.formats

    val responseFuture: Future[HttpResponse] =
      Http(system).singleRequest(HttpRequest(uri = statisticsServiceUrl))

    responseFuture flatMap  { response =>
      Unmarshal(response.entity).to[StatisticsResponse] map { statisticsResponse =>
        statisticsResponse.query.results.quote
      }
    }
  }
} 

//设置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 io.reco

import scala.concurrent.{ExecutionContext, Future}

import akka.actor.ActorSystem
import akka.http.scaladsl.model.Uri.Query
import akka.http.scaladsl.model._
import akka.http.scaladsl.Http
import akka.stream.Materializer
import akka.http.scaladsl.unmarshalling.Unmarshal
import spray.json.DefaultJsonProtocol._
import spray.json._

final case class TokenResult(access_token: String, token_type: String, expires_in: Long)

object TokenResultJsonSupport {
  implicit val tokenFormat = jsonFormat3(TokenResult)
}

class Oauth(conf: Conf) {
  private val AUTH_URI = "/v2.9/dialog/oauth"
  private val AUTH_TOKEN_URI = "/v2.9/oauth/access_token"
  private val params = Map(("client_id", conf.fbId),
    ("redirect_uri", "http://localhost:8080/callback"))
  val uri = Uri(AUTH_URI).withHost(conf.fbUrl).withQuery(Query(params))

  def getToken(code: String)(implicit sys: ActorSystem, mat: Materializer, ec: ExecutionContext): Future[JsValue] = {
    val token_prams = params ++  Map(
      ("client_secret", conf.fbSecret),
      ("code", code))
    val httpRequest = HttpRequest(uri = Uri(AUTH_TOKEN_URI).withHost("graph.facebook.com").withQuery(Query(token_prams)))

    for {
      response <- Http().singleRequest(httpRequest.copy(uri = s"https:${httpRequest.uri}"))
      entity <- Unmarshal(response.entity).to[String]
    } yield entity.parseJson
  }
}

object Oauth {
  def apply(conf: Conf): Oauth = new Oauth(conf)
} 

//设置package包名称以及导入依赖的类
package ko.akka.actors.commands

import akka.actor._
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.{HttpMethods, HttpRequest}
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.stream.ActorMaterializer
import ko.akka.actors.IncomingWebhook.Message
import ko.akka.actors.commands.GoogleActor.Search

object GoogleActor {
  val name : String = "Google"
  val description : String = "search engine"

  case class Search(query: String)

  def props(): Props = {
    Props(classOf[GoogleActor])
  }
}

class GoogleActor extends Actor with ActorLogging{

  import GoogleActor.Search

  import context.dispatcher
  implicit val system = context.system
  implicit val materializer = ActorMaterializer()

  override def receive: Receive = {
    case Search(query) =>
      // 1. ?? ???? ???
      val url: String = s"https://www.google.co.kr/?gws_rd=ssl#newwindow=1&q=$query"
      Http().singleRequest(HttpRequest(uri = url, method = HttpMethods.GET)).map { response =>
        // 2. incomimng Webhook Actor?? ??
        // parent? ??
        Unmarshal(response.entity).to[String].map { responseString =>
          log.info(responseString)
          context.parent ! Message(responseString)
        }
      }


  }
} 

//设置package包名称以及导入依赖的类
package me.alexray.wolfram.impl

import java.net.URLEncoder

import akka.NotUsed
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.{HttpRequest, Uri}
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.stream.Materializer
import akka.util.ByteString
import com.lightbend.lagom.scaladsl.api.ServiceCall
import me.alexray.wolfram.api.WolframService
import play.api.Configuration

import scala.concurrent.{ExecutionContext, Future}


class WolframServiceImpl(config: Configuration)
                        (implicit system: ActorSystem, mat: Materializer, ec: ExecutionContext)
  extends WolframService
{

  val appID = config.underlying.getString("wolfram.appid")
  val apiUrl = s"http://api.wolframalpha.com/v2/"


  override def query(q: String): ServiceCall[NotUsed, String] = ServiceCall { _ =>

    val url = apiUrl + s"query?appid=$appID&input=" + URLEncoder.encode(q, "UTF-8")

    for {
      response <- Http().singleRequest(HttpRequest(uri = Uri(url)))
      if response.status.isSuccess()
      data <- Unmarshal(response).to[String]
    } yield data

  }

  override def simple(q: String): ServiceCall[NotUsed, Array[Byte]] = ServiceCall { _ =>

    println(s"quetions = '$q'")

    val url = apiUrl + s"simple?appid=$appID&input=" +  URLEncoder.encode(q, "UTF-8").replace("+", "%20")

    println(s"url = '$url'")

    for {
      response <- Http().singleRequest(HttpRequest(uri = Uri(url)))
      if response.status.isSuccess()
      bytes <- Unmarshal(response).to[ByteString]
    } yield {
      println(s"received image ${bytes.size} bytes long")
      bytes.toArray
    }

  }
} 

//设置package包名称以及导入依赖的类
import akka.actor.{Actor, ActorSystem, Props}
import akka.http.scaladsl.Http
import akka.http.scaladsl.Http.{IncomingConnection, ServerBinding}
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.http.scaladsl.unmarshalling.PredefinedFromEntityUnmarshallers._
import akka.http.scaladsl.model.HttpMethods._
import akka.http.scaladsl.model._
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Sink, Source}
import org.apache.spark.streaming.receiver.ActorHelper
import org.apache.spark._
import org.apache.spark.streaming._
import scala.concurrent.Future

class ForwarderActor extends Actor with ActorHelper {
  def receive = {
    case data: String => store(data)
  }
}

object Main extends App {
  implicit val system = ActorSystem("sparkDriverActorSystem")
  implicit val mat = ActorMaterializer()
  implicit val ec = system.dispatcher

  val host = "localhost"
  val restPort = 9090
  val actorName = "forwarder"
  val sparkDriverPort = 7777

  // Spark Streaming
  val conf = new SparkConf().setMaster("local[*]").setAppName("TestWebApp").set("spark.driver.port", sparkDriverPort.toString).set("spark.driver.host", host).set("spark.akka.heartbeat.interval", "1s")
  val ssc = new StreamingContext(conf, Seconds(30))
  ssc.actorStream[String](Props[ForwarderActor], actorName).print()

  // Akka HTTP
  val restSource: Source[IncomingConnection, Future[ServerBinding]] = Http().bind(interface = host, port = restPort)

  val handler: HttpRequest => HttpResponse = {
    case HttpRequest(GET, Uri.Path("/"), _, _, _) =>
      HttpResponse(entity = HttpEntity(ContentTypes.`text/html(UTF-8)`, "<html><body>Hi there!</body></html>"))

    case HttpRequest(POST, Uri.Path("/data"), _, entity, _) if entity.contentType == ContentTypes.`application/json` => {
      val url: String = s"akka.tcp://[email protected]$host:${sparkDriverPort + 1}/user/Supervisor0/$actorName"
      val s: Future[String] = Unmarshal(entity).to[String]
      s foreach (system.actorSelection(url) ! _)

      HttpResponse(200)
    }

    case _: HttpRequest =>
      HttpResponse(404, entity = "Page not found!")
  }

  // Start all the things
  ssc.start()

  val binding: Future[ServerBinding] = restSource.to(Sink.foreach { _ handleWithSyncHandler handler }).run()
} 

//设置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 rest.client

import akka.http.scaladsl.model.{HttpResponse, StatusCodes}
import akka.http.scaladsl.unmarshalling.{FromEntityUnmarshaller, Unmarshal}
import akka.stream.ActorMaterializer
import rest.client.entities.ExecutionResultCode

import scala.concurrent.{ExecutionContext, Future}

trait EntityUnmarshallers {

  implicit val mat: ActorMaterializer
  implicit val ec: ExecutionContext

  def unmarshalEntityTo[A](response: HttpResponse)(implicit u: FromEntityUnmarshaller[A]): Future[A] = {
    response.status match {
      case StatusCodes.OK => Unmarshal(response.entity).to[A]
      case _ =>
        val error = new Exception(response.toString())
        response.discardEntityBytes()
        Future.failed(error)
    }
  }

  def extractStatusCode(response: HttpResponse): Future[Int] = {
    val eventualInt = response.status match {
      case StatusCodes.OK => Future.successful(ExecutionResultCode.OK)
      case _              => Future.failed(new Exception(response.toString()))
    }
    response.discardEntityBytes()
    eventualInt
  }

} 

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

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.HttpRequest
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.stream.ActorMaterializer
import com.kimstebel.alexa.AccountResponses.{Readings, Reading}
import de.heikoseeberger.akkahttpcirce.CirceSupport
import io.circe.generic.auto._

import scala.concurrent.{ExecutionContext, Future}

final case class AccountRequests(accountId: String)(
    implicit context: ExecutionContext,
    system: ActorSystem,
    materializer: ActorMaterializer
) extends CirceSupport {

  private[this] def accountUrl(relativePath: String) =
    s"http://meters.uat.ptl.ovotech.org.uk/accounts/$accountId$relativePath?source=DTMF"

  def fetchReadings(): Future[List[Readings]] =
    Http()
      .singleRequest(HttpRequest(uri = accountUrl("/readings")))
      .flatMap(response ? Unmarshal(response).to[List[Readings]])

  def fetchLatestElectricityReading(): Future[Option[Reading]] =
    fetchReadings().map { readings ?
      readings
        .find(_.chargeItem == "Electricity")
        .flatMap(_.readings.find(_.actual))
    }
} 

//设置package包名称以及导入依赖的类
package com.github.notyy.client

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
import akka.http.scaladsl.model.Uri.{Path, Query}
import akka.http.scaladsl.model._
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Sink, Source}
import com.github.notyy.domain.Product
import com.typesafe.scalalogging.slf4j.StrictLogging
import spray.json.{DefaultJsonProtocol, RootJsonFormat}

import scala.concurrent.{ExecutionContextExecutor, Future}

trait ProductSerializer extends SprayJsonSupport with DefaultJsonProtocol with StrictLogging {
  implicit val ProductFormat: RootJsonFormat[Product] = jsonFormat2(Product)
}

trait ProductClient extends ProductSerializer {
  protected def host: String

  protected def port: Int

  implicit val system = ActorSystem("my-system")
  implicit val materializer = ActorMaterializer()
  implicit val executionContext: ExecutionContextExecutor = system.dispatcher

  def queryProductByName(name: String): Future[Option[Product]] = {
    val theUri = Uri(path = Path("/product/")).withQuery(Query(Map("name" -> s"$name")))
    val source = Source.single(HttpRequest(uri = theUri))
    val flow = Http().outgoingConnection(host, port).mapAsync(1) { rs =>
      rs.status match {
        case StatusCodes.OK => {
          Unmarshal(rs.entity).to[Product].map(Some(_))
        }
        case StatusCodes.NotFound => Future(None)
        case _ => Future(None)
      }
    }
    source.via(flow).runWith(Sink.head)
  }
} 

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

import akka.http.scaladsl.marshalling.Marshaller.withFixedContentType
import akka.http.scaladsl.marshalling.ToEntityMarshaller
import akka.http.scaladsl.model.{ContentTypes, HttpEntity, HttpRequest}
import akka.http.scaladsl.unmarshalling.{FromRequestUnmarshaller, Unmarshal, Unmarshaller}
import akka.stream.Materializer
import com.fasterxml.jackson.annotation.JsonInclude.Include
import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper, PropertyNamingStrategy}
import com.fasterxml.jackson.module.scala.DefaultScalaModule

import scala.concurrent.ExecutionContext


package object json {
  val objectMapper = new ObjectMapper()
    .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
    .setSerializationInclusion(Include.NON_EMPTY)
    .setPropertyNamingStrategy(PropertyNamingStrategy.SNAKE_CASE)
    .registerModule(DefaultScalaModule)

  implicit class ObjAsJsonUsingJackson(obj: Any) {
    def asJson: String = objectMapper.writeValueAsString(obj)
  }

  implicit class StringJsonAsCaseClass(json: String) {
    def asObject[T](implicit m: Manifest[T]): T = objectMapper.readValue(json, m.runtimeClass).asInstanceOf[T]
  }

  implicit def jsonMarshaller[T]: ToEntityMarshaller[T] =
    withFixedContentType(ContentTypes.`application/json`) { any =>
      HttpEntity(ContentTypes.`application/json`, any.asJson)
    }

  implicit def jsonUnmarshaller[T](implicit m: Manifest[T], materializer: Materializer): FromRequestUnmarshaller[T] =
    Unmarshaller[HttpRequest, T] {
      implicit ec: ExecutionContext => r => Unmarshal(r.entity).to[String].map(_.asObject[T])
    }
} 

//设置package包名称以及导入依赖的类
package io.vamp.container_driver.marathon

import akka.NotUsed
import akka.actor.Actor
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.HttpHeader.ParsingResult
import akka.http.scaladsl.model.{ HttpHeader, HttpRequest, Uri }
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Source
import de.heikoseeberger.akkasse.{ EventStreamUnmarshalling, ServerSentEvent }
import io.vamp.common.akka.{ CommonActorLogging, ExecutionContextProvider }
import io.vamp.container_driver.ContainerChangeEvent
import org.json4s.native.JsonMethods._
import org.json4s.{ DefaultFormats, StringInput }

trait MarathonSse {
  this: Actor with CommonActorLogging with ExecutionContextProvider ?

  import EventStreamUnmarshalling._

  private implicit val formats = DefaultFormats

  def openEventStream(uri: Uri): Unit = {
    implicit val actorMaterializer = ActorMaterializer()(context)
    Source.single(HttpRequest(uri = "/v2/events", headers = List(HttpHeader.parse("Accept", "text/event-stream").asInstanceOf[ParsingResult.Ok].header)))
      .via(Http()(context.system).outgoingConnection(uri.authority.host.address, uri.authority.port))
      .mapAsync(1)(Unmarshal(_).to[Source[ServerSentEvent, NotUsed]])
      .runForeach(_.runForeach { e ?
        e.`type`.foreach(t ? e.data.foreach(d ? onEvent(t ? d)))
      })
  }

  private def onEvent: PartialFunction[(String, String), Unit] = {
    case (t, data) if t == "deployment_step_success" ?
      val ids = (parse(StringInput(data), useBigDecimalForDouble = true) \ "plan" \ "steps" \\ "actions" \ "app").extract[List[String]]
      ids.foreach { id ?
        log.info(s"marathon deployment event for: '$id'")
        self ! ContainerChangeEvent(id)
      }
  }
} 

//设置package包名称以及导入依赖的类
package com.stulsoft.pakka.http.client.simple

import akka.actor.ActorSystem
import com.typesafe.scalalogging.LazyLogging
import akka.http.scaladsl.Http
import akka.http.scaladsl.model._
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.stream.ActorMaterializer

import scala.concurrent.{Await, Future}
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global


object SimpleClient1 extends App with LazyLogging {
  logger.info("Started")
  implicit val system = ActorSystem()
  implicit val materializer = ActorMaterializer()

  val responseFuture: Future[HttpResponse] =
    Http().singleRequest(HttpRequest(uri = "http://akka.io"))

  val result = Await.result(responseFuture, 5.seconds)

  logger.debug("Status is {}", result.status)
  logger.debug("Content type is {}", result.entity.contentType)

  val resultAsString = Await.result(Unmarshal(result.entity).to[String], 5.seconds)
  println(resultAsString)

  logger.info("Finished")

  materializer.shutdown()
  sys.exit(0)
} 

//设置package包名称以及导入依赖的类
import akka.actor.ActorSystem

import akka.event.{LoggingAdapter, Logging}
import akka.http.scaladsl.Http
import akka.http.scaladsl.client.RequestBuilding
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
import akka.http.scaladsl.marshalling.ToResponseMarshallable
import akka.http.scaladsl.model.{HttpResponse, HttpRequest}
import akka.http.scaladsl.model.StatusCodes._
import akka.http.scaladsl.server.Directives._
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.stream.{ActorMaterializer, Materializer}
import akka.stream.scaladsl.{Flow, Sink, Source}
import com.typesafe.config.Config
import com.typesafe.config.ConfigFactory
import java.io.IOException
import scala.concurrent.{ExecutionContextExecutor, Future}
import scala.math._
import spray.json.DefaultJsonProtocol


package util.http {

  class HttpDispatcher { 

   implicit val system = ActorSystem()
   implicit val executor = system.dispatcher
   implicit val materializer = ActorMaterializer()

    lazy val setlisterConnectionFlow: Flow[HttpRequest, HttpResponse, Any] = Http().outgoingConnection("www.setlister.me", 80)  

    def setlisterRequest(request: HttpRequest): Future[HttpResponse] = Source.single(request).via(setlisterConnectionFlow).runWith(Sink.head)

    def getURL(url: String) {
      setlisterRequest(RequestBuilding.Get(url)).flatMap { response =>
        response.status match {
          case OK =>  {
            Unmarshal(response.entity).to[String].flatMap { entity =>
              Future.successful(println(entity))
            }
          }
          case BadRequest => Future.successful(Left(s"incorrect IP format"))
          case _ => Unmarshal(response.entity).to[String].flatMap { entity =>
            val error = s"FreeGeoIP request failed with status code ${response.status} and entity $entity"
            //logger.error(error)
            println("error")
            Future.failed(new IOException(error))
          }
        }
      }
    }
  }
} 

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

import java.io.IOException

import scala.concurrent.Future

import akka.http.scaladsl.Http
import akka.http.scaladsl.client.RequestBuilding
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
import akka.http.scaladsl.marshalling.ToResponseMarshallable
import akka.http.scaladsl.model.{ HttpResponse, HttpRequest }
import akka.http.scaladsl.model.StatusCodes.{ OK, BadRequest }
import akka.http.scaladsl.model.Uri
import akka.http.scaladsl.server.Directives
import akka.http.scaladsl.server.Route
import akka.http.scaladsl.unmarshalling.Unmarshal

import spray.json._

import org.yashsriv.helpers.Worker
import org.yashsriv.json.MovieSupport
import org.yashsriv.models.MovieQuery
import org.yashsriv.models.MovieResult

trait Movie extends Directives with Worker with MovieSupport with SprayJsonSupport {

  def omdbApiRequest(request: HttpRequest): Future[HttpResponse] = Http().singleRequest(request)

  def movieApi(implicit requestUri: Uri): Route = pathPrefix("movie") {
    (post & entity(as[MovieQuery])) { movieQuery ?
      complete {
        fetchMovieInfo(movieQuery).map[ToResponseMarshallable] {
          case Right(movieInfo) => movieInfo
          case Left(errorMessage) => BadRequest ? errorMessage
        }
      }
    }
  }

  def fetchMovieInfo(mq: MovieQuery)(implicit requestUri: Uri): Future[Either[String, MovieResult]] = {
    omdbApiRequest(RequestBuilding.Get(requestUri withQuery convertToQuery(mq))).flatMap { response =>
      response.status match {
        case OK         ? Unmarshal(response.entity).to[MovieResult].map(Right(_))
        case BadRequest ? Future.successful(Left(s"${mq.toJson.prettyPrint} \nIncorrect Movie Format"))
        case _          ? Unmarshal(response.entity).to[String].flatMap { entity ?
          val error = s"Omdb request failed with status code ${response.status} and entity $entity"
          Future.failed(new IOException(error))
        }
      }
    }
  }

} 

//设置package包名称以及导入依赖的类
package ch.becompany.http

import java.io.IOException

import akka.http.scaladsl.model.StatusCodes.{NotModified, OK}
import akka.http.scaladsl.model._
import akka.http.scaladsl.model.headers.EntityTag
import akka.http.scaladsl.unmarshalling.Unmarshal

import scala.concurrent.{ExecutionContext, Future}
import scalacache._
import scalacache.caffeine._

trait CachingSupport extends HttpClient {

  type Cached[A] = (EntityTag, A)
  implicit val etagCache = ScalaCache(CaffeineCache())

  override def handle[A](handler: HttpHandler[A], request: HttpRequest, response: HttpResponse)
                        (implicit ec: ExecutionContext): Future[A] =
    response.status match {
      case OK =>
        for {
          body <- super.handle(handler, request, response)
          cached <- addToCache(request.uri, response, body)
        } yield body
      case NotModified => readFromCache(request.uri)
      case _ => super.handle(handler, request, response)
    }

  abstract override def additionalHeaders(req: HttpRequest)(implicit ec: ExecutionContext): Future[Seq[HttpHeader]] =
    get[Cached[_], NoSerialization](req.uri) map {
      case Some((etag, _)) => scala.collection.immutable.Seq(headers.`If-None-Match`(etag))
      case None => Nil
    }

  private def addToCache[A](uri: Uri, response: HttpResponse, body: A)(implicit ec: ExecutionContext): Future[Unit] =
    response.header[headers.ETag].map(_.etag).map { etag =>
      put(uri)((etag, body))
    } getOrElse Future()

  private def addToCacheAndRead(uri: Uri, response: HttpResponse)(implicit ec: ExecutionContext): Future[ResponseEntity] =
    response.header[headers.ETag].map(_.etag).map { etag =>
      Unmarshal(response.entity).to[String].
        flatMap(content => put(uri)((etag, content))).
        flatMap(_ => readFromCache(uri))
    } getOrElse Future(response.entity)

  private def readFromCache[A](uri: Uri)(implicit ec: ExecutionContext): Future[A] = {
    get[Cached[A], NoSerialization](uri) flatMap {
      case Some((etag, content)) => Future(content)
      case None => Future.failed(new IOException("Cache entry not found"))
    }
  }

} 

//设置package包名称以及导入依赖的类
package ch.becompany.http

import java.io.IOException

import akka.http.scaladsl.model.StatusCodes.OK
import akka.http.scaladsl.model.{HttpRequest, HttpResponse, ResponseEntity}
import akka.http.scaladsl.unmarshalling.{Unmarshal, Unmarshaller}
import akka.stream.ActorMaterializer
import com.typesafe.scalalogging.LazyLogging

import scala.concurrent.{ExecutionContext, Future}

trait HttpHandler[A] {
  def handle(request: HttpRequest, response: HttpResponse)(implicit ec: ExecutionContext): Future[A]
}

class UnmarshallingHttpHandler[A](implicit materializer: ActorMaterializer, unmarshaller: Unmarshaller[ResponseEntity, A])
  extends HttpHandler[A] with LazyLogging{

  def handle(request: HttpRequest, response: HttpResponse)(implicit ec: ExecutionContext): Future[A] =
    response.status match {
      case OK => Unmarshal(response.entity).to[A]
      case _ => handleError(response)
    }

  def handleError(response: HttpResponse)(implicit ec: ExecutionContext): Future[A] = {
    Unmarshal(response.entity).to[String].flatMap { entity =>
      val error = s"HTTP error ${response.status}: $entity"
      logger.error(error)
      Future.failed(new IOException(error))
    }
  }

} 

//设置package包名称以及导入依赖的类
package ch.becompany.http.oauth

import akka.http.scaladsl.model.headers.GenericHttpCredentials
import akka.http.scaladsl.model.{HttpHeader, HttpRequest, headers}
import akka.http.scaladsl.unmarshalling.Unmarshal
import ch.becompany.http.HttpClient
import com.hunorkovacs.koauth.domain.KoauthRequest
import com.hunorkovacs.koauth.service.consumer.DefaultConsumerService

import scala.concurrent.{ExecutionContext, Future}

trait OAuthSupport extends HttpClient {

  private val oauthHeaderName = "OAuth"

  val oauthConfig: OAuthConfig

  private lazy val consumer = new DefaultConsumerService(system.dispatcher)

  def oauthHeader(req: HttpRequest)(implicit ex: ExecutionContext): Future[HttpHeader] =
    for {
      body: Option[String] <-
        if (req.entity.isKnownEmpty()) Future(None)
        else Unmarshal(req.entity).to[String].map(Some(_))
      header <- oauthHeader(req, body)
    } yield header

  def oauthHeader(request: HttpRequest, body: Option[String])(implicit ex: ExecutionContext): Future[HttpHeader] =
    consumer.createOauthenticatedRequest(
      KoauthRequest(
        method = request.method.value,
        url = request.uri.toString,
        authorizationHeader = None,
        body = body
      ),
      oauthConfig.consumerKey,
      oauthConfig.consumerSecret,
      oauthConfig.accessToken,
      oauthConfig.accessTokenSecret
    ).
      map(_.header).
      map(header => headers.Authorization(GenericHttpCredentials(
        oauthHeaderName, header.substring(oauthHeaderName.length + 1))))

  abstract override def additionalHeaders(request: HttpRequest)(implicit ec: ExecutionContext): Future[Seq[HttpHeader]] = {
    val headersFuture = super.additionalHeaders(request)
    val oauthHeaderFuture = oauthHeader(request)
    for (headers <- headersFuture; oauthHeader <- oauthHeaderFuture)
      yield headers :+ oauthHeader
  }

} 

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

import actors.WebSocketActor.Completed
import akka.actor.{Actor, ActorLogging, ActorRef, Props}
import akka.http.scaladsl.Http
import akka.http.scaladsl.client.RequestBuilding.Get
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Sink, Source}
import de.heikoseeberger.akkasse.EventStreamUnmarshalling._
import de.heikoseeberger.akkasse.ServerSentEvent

import scala.util.{Failure, Success}


object WebSocketActor {
    def props(out: ActorRef, mat: ActorMaterializer) = Props(new WebSocketActor(out, mat))

    case class Completed()
}

class WebSocketActor(out: ActorRef, mat: ActorMaterializer) extends Actor with ActorLogging {
    implicit val m = mat
    implicit val s = context.system
    implicit val d = context.system.dispatcher

    def receive = {
        case msg: String =>
            out ! s"echoing: $msg"
    }

    override def preStart(): Unit = {
        log.warning("starting stream")
        out ! "starting stream"
        val f = sseSource().runForeach(_.map { sse => println(sse); sse.data }.runWith(Sink.actorRef(out, Completed())))
        f.onComplete {
            case Success(_) => out ! "complete"
            case Failure(ex) => out ! ex.getMessage
        }
    }

    def sseSource() = {
        Source.single(Get())
        .via(Http().outgoingConnection("127.0.0.1", 9001))
        .mapAsync(1)(Unmarshal(_).to[Source[ServerSentEvent, Any]])
    }
}