//设置package包名称以及导入依赖的类
package bot.line.client

import akka.http.scaladsl.client.RequestBuilding
import akka.http.scaladsl.model.headers.OAuth2BearerToken
import akka.http.scaladsl.model._
import bot.line.json.MessagesJsonSupport
import bot.line.model.send.{Messages, TextMessage}

import scala.concurrent.ExecutionContext

case class ReplyMessageRequest(
                                accessToken: String,
                                replyToken: String,
                                message: String
                              ) extends MessagesJsonSupport {

  def httpRequest(implicit ec: ExecutionContext): HttpRequest = {
    val auth:HttpHeader = headers.Authorization(OAuth2BearerToken(accessToken))
    val content = Messages(
      replyToken = replyToken,
      messages = List(TextMessage(text = message))
    )
    RequestBuilding.Post(
      uri = "https://api.line.me/v2/bot/message/reply",
      content = content
    ).withHeaders(auth)
  }
} 

//设置package包名称以及导入依赖的类
package au.csiro.data61.magda.external

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.client.RequestBuilding
import akka.http.scaladsl.model._
import akka.stream.Materializer
import akka.stream.scaladsl._
import au.csiro.data61.magda.util.Http.getPort

import scala.concurrent.{ ExecutionContext, Future }

class HttpFetcher(interfaceConfig: InterfaceConfig, implicit val system: ActorSystem,
  implicit val materializer: Materializer, implicit val ec: ExecutionContext) {

  lazy val connectionFlow: Flow[HttpRequest, HttpResponse, Any] =
    Http().outgoingConnection(interfaceConfig.baseUrl.getHost, getPort(interfaceConfig.baseUrl))

  def request(path: String): Future[HttpResponse] =
    interfaceConfig.fakeConfig match {
      case Some(fakeConfig) => Future {
        val file = io.Source.fromInputStream(getClass.getResourceAsStream(fakeConfig.datasetPath))

        val response = new HttpResponse(
          status = StatusCodes.OK,
          headers = scala.collection.immutable.Seq(),
          protocol = HttpProtocols.`HTTP/1.1`,
          entity = HttpEntity(ContentTypes.`application/json`, file.mkString))

        file.close()

        response
      }
      case None => {
        val request = RequestBuilding.Get(s"${interfaceConfig.baseUrl.getPath}${path}")
        Source.single(request).via(connectionFlow).runWith(Sink.head)
      }
    }
} 

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

import akka.http.scaladsl.client.RequestBuilding
import akka.http.scaladsl.marshalling.ToEntityMarshaller
import akka.http.scaladsl.model.{ HttpMethods, HttpRequest }
import hmda.api.headers.{ HmdaInstitutionsHeader, HmdaUsernameHeader }

trait RequestHeaderUtils extends RequestBuilding {
  import HttpMethods._

  def getWithCfpbHeaders(path: String): HttpRequest = {
    new RequestBuilder(GET).apply(path)
      .addHeader(usernameHeader)
      .addHeader(institutionsHeader)
  }

  def postWithCfpbHeaders[T, ec: EC](path: String, content: T)(implicit m: ToEntityMarshaller[T]) = {
    new RequestBuilder(POST).apply(path, content)
      .addHeader(usernameHeader)
      .addHeader(institutionsHeader)
  }

  def postWithCfpbHeaders(path: String) = {
    new RequestBuilder(POST).apply(path)
      .addHeader(usernameHeader)
      .addHeader(institutionsHeader)
  }

  val usernameHeader = new HmdaUsernameHeader("banker11")
  val institutionsHeader = new HmdaInstitutionsHeader(List("0", "xxxxx"))

} 

//设置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 xyz.joaovasques.sparkapi.api.standalone.interactors

import xyz.joaovasques.sparkapi.messages.SparkApiMessages._
import scala.concurrent.Future
import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
import akka.actor.ActorSystem
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.http.scaladsl.model.ResponseEntity
import akka.http.scaladsl.client.RequestBuilding
import xyz.joaovasques.sparkapi.helpers._
import akka.stream.ActorMaterializer
import akka.http.scaladsl.model.StatusCodes._
import xyz.joaovasques.sparkapi.actors.SparkActor._

class CheckJobStatusInteractor(val sparkApi: HttpRequest => Future[HttpResponse],
  val master: String, val apiVersion: String = "v1"
)(implicit system: ActorSystem) extends Interactor[String, SparkJobStatusResponse] with JsonHelpers {
  private val endpoint = s"/${apiVersion}/submissions/status"
  implicit val executionContext = system.dispatcher
  implicit val materializer = ActorMaterializer()

  private val unmarshaller: ResponseEntity => Future[SparkJobStatusResponse] =
  { entity => Unmarshal(entity).to[SparkJobStatusResponse] }

  private def handleSparkResponse(response: HttpResponse): Future[SparkJobStatusResponse] =
    response.status match {
      case OK => unmarshaller(response.entity)
      case otherStatus =>
        throw new Exception(s"Error Communicating with Spark. Status code ${otherStatus}")
    }

  def call(driverId: String): Future[SparkJobStatusResponse] = {
    val httpRequest = RequestBuilding.Get(s"${endpoint}/${driverId}")
    sparkApi(httpRequest) flatMap { handleSparkResponse(_) }
  }
} 

//设置package包名称以及导入依赖的类
package xyz.joaovasques.sparkapi.api.standalone.interactors

import scala.concurrent.Future

import akka.actor.ActorSystem
import akka.http.scaladsl.client.RequestBuilding
import akka.http.scaladsl.model.{ContentTypes, HttpEntity, HttpRequest, HttpResponse, ResponseEntity}
import akka.http.scaladsl.model.StatusCodes._
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.stream.ActorMaterializer
import xyz.joaovasques.sparkapi.helpers._
import xyz.joaovasques.sparkapi.messages.SparkApiMessages._
import xyz.joaovasques.sparkapi.actors.SparkActor._

class KillJobInteractor(val sparkApi: HttpRequest => Future[HttpResponse],
  val master: String, val apiVersion: String = "v1"
)(implicit system: ActorSystem) extends Interactor[String, SparkJobKillResponse] with JsonHelpers {

  private val endpoint = s"/${apiVersion}/submissions/kill"
  implicit val executionContext = system.dispatcher
  implicit val materializer = ActorMaterializer()

  private val unmarshaller: ResponseEntity => Future[SparkJobKillResponse] =
  { entity => Unmarshal(entity).to[SparkJobKillResponse] }

  private def handleSparkResponse(response: HttpResponse): Future[SparkJobKillResponse] =
    response.status match {
      case OK => unmarshaller(response.entity)
      case otherStatus =>
        Future.failed(new Exception(s"Communicating with Spark: status code ${otherStatus}"))
    }

  def call(driverId: String): Future[SparkJobKillResponse] = {
    val httpRequest = RequestBuilding.Post(s"${endpoint}/${driverId}", HttpEntity(ContentTypes.`application/json`, ""))
      .addHeader(header)

    sparkApi(httpRequest) flatMap { handleSparkResponse(_) }
  }
} 

//设置package包名称以及导入依赖的类
package xyz.joaovasques.sparkapi.api.standalone.interactors

import xyz.joaovasques.sparkapi.messages.SparkApiMessages._
import scala.concurrent.Future
import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
import akka.actor.ActorSystem
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.http.scaladsl.model.{ HttpHeader, ResponseEntity }
import akka.http.scaladsl.model.HttpHeader.ParsingResult
import akka.http.scaladsl.client.RequestBuilding
import akka.http.scaladsl.model.{ ContentTypes, HttpEntity }
import xyz.joaovasques.sparkapi.helpers._
import akka.stream.ActorMaterializer
import akka.http.scaladsl.model.StatusCodes._
import xyz.joaovasques.sparkapi.exceptions.SparkApiExceptions._
import xyz.joaovasques.sparkapi.actors.SparkActor._

class SubmitJobInteractor(val sparkApi: HttpRequest => Future[HttpResponse],
  val master: String, val apiVersion: String = "v1"
)(implicit system: ActorSystem) extends Interactor[SubmitJob, SparkJobSumissionResponse] with JsonHelpers {

  private val endpoint = s"/${apiVersion}/submissions/create"
  implicit val executionContext = system.dispatcher
  implicit val materializer = ActorMaterializer()

  private val unmarshaller: ResponseEntity => Future[SparkJobSumissionResponse] =
  { entity => Unmarshal(entity).to[SparkJobSumissionResponse] }

  private def handleSparkResponse(response: HttpResponse): Future[SparkJobSumissionResponse] =
    response.status match {
      case OK =>
        for(unmarshalledResponse <- unmarshaller(response.entity)) yield {
          if(unmarshalledResponse.success)
            unmarshalledResponse
          else
          throw JobSubmissionFailedException()
        }
      case otherStatus =>
        throw new SparkCommunicationException(otherStatus.intValue)
    }

  def call(request: SubmitJob): Future[SparkJobSumissionResponse] = {
    val httpRequest = RequestBuilding.Post(endpoint,
      HttpEntity(ContentTypes.`application/json`, generateSubmitJsonBody(request, master, request.envVars)))
      .addHeader(header)

    sparkApi(httpRequest) flatMap { handleSparkResponse(_) }
  }
} 

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

import java.io.IOException

import akka.actor.ActorSystem
import akka.event.Logging
import akka.http.scaladsl.Http
import akka.http.scaladsl.client.RequestBuilding
import akka.http.scaladsl.model.StatusCodes._
import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Flow, Sink, Source}
import com.typesafe.config.ConfigFactory

import scala.concurrent.Future



object Starter extends App {

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

  val config = ConfigFactory.load()
  val logger = Logging(system, getClass)

  lazy val apiFlow: Flow[HttpRequest, HttpResponse, Any] =
    Http().outgoingConnectionHttps(config.getString("services.trade-api.host"), config.getInt("services.trade-api.port"))

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

  def fetchInfo(symbol: String): Future[Either[String, String]] = {
    val apiurl = config.getString("services.trade-api.trade")
    apiRequest(RequestBuilding.Get(apiurl+"?symbol="+symbol)).flatMap { response =>
      response.status match {
        case OK =>
          Unmarshal(response.entity).to[String].map(Right(_))
        case BadRequest =>
          Future.successful(Left(s"$symbol: incorrect Symbol format"))
        case _ =>
          Unmarshal(response.entity).to[String].flatMap { entity =>
            val error = s"Trade request failed with status code ${response.status} and entity $entity"
            logger.error(error)
            Future.failed(new IOException(error))
          }
      }
    }
  }

  val res = fetchInfo("btc_cny").map {
    case Right(info) => logger.info("Result: {}", info)
    case Left(errorMessage) => logger.info("Result: {}", errorMessage)
  }
}