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

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.HttpRequest
import akka.stream.ActorMaterializer
import com.github.chaabaj.openid.exceptions.{MalformedResponseException, WebServiceException}
import com.github.chaabaj.openid.utils.JsonResponseParser
import spray.json.JsValue

import scala.concurrent.duration.FiniteDuration
import scala.concurrent.{ExecutionContext, Future}
import scala.util.{Failure, Success}

private [openid] class HttpClient(implicit actorSystem: ActorSystem, timeout: FiniteDuration) {
  implicit val materializer = ActorMaterializer()
  val responseParser = new JsonResponseParser
  val http = Http()

  def request(httpRequest: HttpRequest)(implicit exc: ExecutionContext): Future[JsValue] =
    for {
      response <- http.singleRequest(httpRequest)
      body <- response.entity.toStrict(timeout).map(_.data.utf8String)
      data <- {
        responseParser.parse(body) match {
          case Success(data) =>
            if (response.status.isFailure()) {
              Future.failed(WebServiceException(response.status, data))
            } else {
              Future.successful(data)
            }
          case Failure(ex) => Future.failed(MalformedResponseException(response.status, ex.toString))
        }
      }
    } yield data
}

private[openid] object HttpClient {
  def apply()(implicit system: ActorSystem, _timeout: FiniteDuration): HttpClient =
    new HttpClient()
} 

//设置package包名称以及导入依赖的类
package org.danielwojda.obfuscator.functionaltests

import akka.http.scaladsl.Http
import akka.http.scaladsl.model.{HttpRequest, HttpResponse, StatusCodes}
import org.danielwojda.obfuscator.functionaltests.dsl.AkkaImplicits
import org.danielwojda.obfuscator.functionaltests.dsl.HttpEntityImplicitConverters._
import org.scalatest.concurrent.PatienceConfiguration.Timeout
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.{FlatSpec, GivenWhenThen, Matchers}

import scala.concurrent.Future
import scala.concurrent.duration._
import scala.language.postfixOps

class MonitoringEndpointsSpec extends FlatSpec with Matchers with GivenWhenThen with ScalaFutures with AkkaImplicits {

  override implicit val patienceConfig = PatienceConfig(timeout = 2 seconds)
  implicit val timeout = Timeout(patienceConfig.timeout)

  "Service" should "expose the 'status' endpoint which always returns OK" in {
    When("a 'status' endpoint is called")
    val responseFuture: Future[HttpResponse] = Http().singleRequest(HttpRequest(uri = "http://localhost:8080/private/status"))

    Then("Http status 200 and OK body is returned")
    whenReady(responseFuture) { response =>
      response.status shouldBe StatusCodes.OK
      response.entity.asString shouldBe "OK"
    }
  }

} 

//设置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 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 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 akka.stream.alpakka.s3.auth

import java.net.URLEncoder

import akka.http.scaladsl.model.Uri.{Path, Query}
import akka.http.scaladsl.model.{HttpHeader, HttpRequest}

// Documentation: http://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html
private[alpakka] case class CanonicalRequest(method: String,
                                             uri: String,
                                             queryString: String,
                                             headerString: String,
                                             signedHeaders: String,
                                             hashedPayload: String) {
  def canonicalString: String = s"$method\n$uri\n$queryString\n$headerString\n\n$signedHeaders\n$hashedPayload"
}

private[alpakka] object CanonicalRequest {
  def from(req: HttpRequest): CanonicalRequest = {
    val hashedBody = req.headers.find(_.name == "x-amz-content-sha256").map(_.value).getOrElse("")
    CanonicalRequest(
      req.method.value,
      pathEncode(req.uri.path),
      canonicalQueryString(req.uri.query()),
      canonicalHeaderString(req.headers),
      signedHeadersString(req.headers),
      hashedBody
    )
  }

  def canonicalQueryString(query: Query): String =
    query.sortBy(_._1).map { case (a, b) => s"${uriEncode(a)}=${uriEncode(b)}" }.mkString("&")

  private def uriEncode(str: String) = URLEncoder.encode(str, "utf-8")

  def canonicalHeaderString(headers: Seq[HttpHeader]): String = {
    val grouped = headers.groupBy(_.lowercaseName())
    val combined = grouped.mapValues(_.map(_.value.replaceAll("\\s+", " ").trim).mkString(","))
    combined.toList.sortBy(_._1).map { case (k, v) => s"$k:$v" }.mkString("\n")
  }

  def signedHeadersString(headers: Seq[HttpHeader]): String =
    headers.map(_.lowercaseName()).distinct.sorted.mkString(";")

  private def pathEncode(path: Path): String =
    if (path.isEmpty) "/"
    else
      path.toString().flatMap {
        case ch if "!$&'()*+,;:=".contains(ch) => "%" + Integer.toHexString(ch.toInt).toUpperCase
        case other => other.toString
      }

} 

//设置package包名称以及导入依赖的类
package org.zalando.react.nakadi.client.providers

import java.security.SecureRandom
import java.security.cert.X509Certificate
import javax.net.ssl.{SSLContext, TrustManager, X509TrustManager}

import akka.actor.ActorContext
import akka.http.scaladsl.Http.OutgoingConnection
import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
import akka.http.scaladsl.settings.ClientConnectionSettings
import akka.http.scaladsl.{Http, HttpsConnectionContext}
import akka.stream.scaladsl.Flow

import scala.concurrent.Future
import scala.concurrent.duration._


class HttpClientProvider(actorContext: ActorContext,
                         server: String, port: Int,
                         isConnectionSSL: Boolean = false,
                         acceptAnyCertificate: Boolean = false,
                         connectionTimeout: FiniteDuration) {

  val http = Http(actorContext.system)

  private val settings = {
    ClientConnectionSettings
      .apply(actorContext.system)
      .withConnectingTimeout(connectionTimeout)
      .withIdleTimeout(Duration.Inf)
  }

  val connection: Flow[HttpRequest, HttpResponse, Future[OutgoingConnection]] = {

    isConnectionSSL match {
      case true =>
        val sslContext = if (!acceptAnyCertificate) SSLContext.getDefault else {

          val permissiveTrustManager: TrustManager = new X509TrustManager() {
            override def checkClientTrusted(chain: Array[X509Certificate], authType: String): Unit = {}
            override def checkServerTrusted(chain: Array[X509Certificate], authType: String): Unit = {}
            override def getAcceptedIssuers(): Array[X509Certificate] = Array.empty
          }

          val ctx = SSLContext.getInstance("TLS")
          ctx.init(Array.empty, Array(permissiveTrustManager), new SecureRandom())
          ctx
        }
        http.outgoingConnectionHttps(server, port, new HttpsConnectionContext(sslContext), settings = settings)
      case false =>
        http.outgoingConnection(server, port, settings = settings)
    }
  }

} 

//设置package包名称以及导入依赖的类
package edu.goldlok.minio_scala.auth

import akka.http.scaladsl.model.{HttpHeader, HttpRequest}
import akka.http.scaladsl.model.Uri.{Path, Query}

import java.net.URLEncoder



case class CanonicalRequest(method: String, uri: String,
                            queryString: String,
                            headerString: String,
                            signedHeaders: String,
                            hashedPayload: String) {
  def canonicalString: String = {
    s"$method\n$uri\n$queryString\n$headerString\n\n$signedHeaders\n$hashedPayload"
  }
}

object CanonicalRequest {
  private[this] val content_sha256 = "x-amz-content-sha256"
  def from(req: HttpRequest): CanonicalRequest = {
    val hashedBody = req.headers.find(_.name == content_sha256).map(_.value).getOrElse("")
    CanonicalRequest(
      req.method.value,
      preprocessPath(req.uri.path),
      canonicalQueryString(req.uri.query()),
      canonicalHeaderString(req.headers),
      signedHeadersString(req.headers),
      hashedBody
    )
  }

  def canonicalQueryString(query: Query): String = {
    query.sortBy(_._1).map { case (a, b) => s"${uriEncode(a)}=${uriEncode(b)}" }.mkString("&")
  }

  def uriEncode(str: String): String = URLEncoder.encode(str, "utf-8")

  def preprocessPath(path: Path): String = {
    uriEncode(path.toString()).replace(":", "%3A").replace("%2F", "/")
  }

  def canonicalHeaderString(headers: Seq[HttpHeader]): String = {
    val grouped = headers.groupBy(_.lowercaseName())
    val combined = grouped.mapValues(_.map(_.value.replaceAll("\\s+", " ").trim).mkString(","))
    combined.toList.sortBy(_._1).map { case (k, v) => s"$k:$v" }.mkString("\n")
  }

  def signedHeadersString(headers: Seq[HttpHeader]): String = {
    headers.map(_.lowercaseName()).distinct.sorted.mkString(";")
  }
} 

//设置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包名称以及导入依赖的类
package net.hvieira.yeoldeonlinestore.api

import akka.actor.ActorSystem
import akka.http.scaladsl.model.HttpRequest
import akka.http.scaladsl.model.StatusCodes._
import akka.http.scaladsl.server.Route
import net.hvieira.yeoldeonlinestore.actor.store.StoreManager
import net.hvieira.yeoldeonlinestore.test.ServiceIntegrationTest

class StoreAPISpec extends ServiceIntegrationTest {

  // TODO probably want to change this to use the akka testkit probe actors and such or with DI
  private val testActorSystem = ActorSystem("test-system")
  private val availableItems = List(
    Item("item1", 0.05),
    Item("123Item", 1.0),
    Item("Ring of Powaaa", 100000000.0)
  )
  private val storeManRef = testActorSystem.actorOf(StoreManager.props(3, () => availableItems))

  private val route = Route.seal(new StoreAPI(storeManRef).route)

  "The store API" should {

    "allow unauthenticated users to see items on sale" in {
      val request: HttpRequest = Get("/store")

      request ~> route ~> check {
        status shouldBe OK
        handled shouldBe true

        val storeFront = entityAs[StoreFront]
        storeFront.items should contain theSameElementsAs availableItems
      }

    }

  }

} 

//设置package包名称以及导入依赖的类
package se.meldrum.machine

import akka.http.scaladsl.testkit.ScalatestRouteTest
import org.scalatest.{Matchers, WordSpec}
import PostgresTestDb._
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.{HttpEntity, HttpMethods, HttpRequest, MediaTypes}
import akka.stream.ActorMaterializer
import akka.util.ByteString
import se.meldrum.machine.http.RestService
import slick.driver.PostgresDriver.api._

import scala.concurrent.ExecutionContext

trait BaseSpec extends WordSpec with Matchers with ScalatestRouteTest {
  dbProcess.getProcessId
  implicit val db = Database.forConfig("postgres-test")
  implicit val sys = ActorSystem("machine")
  implicit val ec = ExecutionContext
  implicit val mat = ActorMaterializer()
  val restService = new RestService()
  val route = restService.route

  def postRequest(path: String, json: ByteString): HttpRequest =
    HttpRequest(HttpMethods.POST,
      uri = path,
      entity = HttpEntity(MediaTypes.`application/json`, json)
    )


  def userJsonRequest(name: String, pass: String, email: String): ByteString =
    ByteString(
      s"""
       |{
       |    "name":"$name",
       |    "password":"$pass",
       |    "email":"$email"
       |}
        """.stripMargin)


  def createTestUsers(): Seq[HttpRequest] = {
    val userOne = userJsonRequest("testuser", "secret", "[email protected]")
    val userTwo = userJsonRequest("testuser2", "secret", "[email protected]")
    val userThree = userJsonRequest("testuser3", "secret", "[email protected]")

    val requests = Seq(
      postRequest("/v1/user/create", userOne),
      postRequest("/v1/user/create", userTwo),
      postRequest("/v1/user/create", userThree)
    )
    requests
  }
} 

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

import java.nio.file.{ Files, Path }

import akka.stream.scaladsl.FileIO

import scala.concurrent._

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.HttpRequest
import akka.stream.Materializer

import scala.util.{ Success, Failure }

class DownloadManager(implicit system: ActorSystem, materializer: Materializer) {
  implicit val ec: ExecutionContext = system.dispatcher

  val http = Http()

  def download(url: String): Future[(Path, Long)] = {
    val tempFileFuture = createTempFile()
    val responseFuture = http.singleRequest(HttpRequest(uri = url))

    for {
      filePath ? tempFileFuture
      response ? responseFuture
      ioRes ? response.entity.dataBytes.runWith(FileIO.toPath(filePath))
    } yield {
      ioRes.status match {
        case Success(_)     ? (filePath, ioRes.count)
        case Failure(cause) ? throw cause
      }
    }
  }

  // FIXME: dispatcher for this
  private def createTempFile(): Future[Path] = {
    Future {
      blocking {
        Files.createTempFile("ActorDownloadManager", "")
      }
    }
  }
} 

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

import akka.actor.{Actor, Props}
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.HttpMethods.GET
import akka.http.scaladsl.model.{HttpRequest, HttpResponse, Uri}

import scala.concurrent.Future
import scala.io.StdIn


class ConnectionActor extends Actor {

  case object StartConnection

  case class ConnectWebsite(s: String)

  override def receive = {
    case StartConnection => {
      val requestHandler: HttpRequest => Future[HttpResponse] = {
        case HttpRequest(GET, Uri.Path("/pintpin"), _, _, _) => {

          val c = context.system.actorOf(Props[RequestHandler], "child1")

          val a = c ? OneSlash
          a map { p =>
            p
          }
        }
        case HttpRequest(GET, Uri.Path("/ping"), _, _, _) =>
          Future {
            HttpResponse(entity = "PONG!")
          }
        case HttpRequest(GET, Uri.Path("/crash"), _, _, _) =>
          sys.error("BOOM!")

      }


      val bindingFuture = Http().bindAndHandleAsync(requestHandler, "localhost", 8080)
      println(s"Server online at http://localhost:8080/\nPress RETURN to stop...")
      StdIn.readLine() // let it run until user presses return
      bindingFuture
        .flatMap(_.unbind()) // trigger unbinding from the port
        .onComplete(_ => system.shutdown()) // and shutdown when done

    }
  }
} 

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

import akka.actor.ActorSystem
import akka.http.scaladsl.model.{ HttpMethods, HttpRequest, HttpResponse, StatusCodes }
import akka.stream.ActorMaterializer
import com.typesafe.config.{ Config, ConfigFactory }
import org.scalatest.concurrent.{ IntegrationPatience, PatienceConfiguration, ScalaFutures }
import org.scalatest.{ Matchers, WordSpec }
import scala.concurrent.Future

class LayerClientIntegrationSpec extends WordSpec with Matchers with PatienceConfiguration with IntegrationPatience with ScalaFutures {
  val testConfig = ConfigFactory.load("test-application.conf")
  val router = new LayerRouter(testConfig)
  implicit val system = ActorSystem("LayerClientIntegrationSpecSystem")
  implicit val materializer = ActorMaterializer()
  implicit val ec = system.dispatcher

  class IntegrationLayerClient(router: LayerRouter, config: Config) extends LayerClient(router, config) {
    def testRequest(httpRequest: HttpRequest): Future[HttpResponse] = executeRequest(httpRequest)
  }

  "#executeRequest" should {
    "complete the request" in {
      val request = HttpRequest(HttpMethods.GET, uri = "https://layer.com")
      val client = new IntegrationLayerClient(router, testConfig)
      client.testRequest(request).futureValue.status shouldBe StatusCodes.OK
    }
  }

} 

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

import akka.actor.{Actor, ActorLogging}
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.HttpRequest
import akka.stream.{ActorMaterializer, ActorMaterializerSettings}

import scala.concurrent.Future
import scala.concurrent.duration._
import scala.language.postfixOps


trait GymClient extends Actor with ActorLogging with JsonSupport {

  import context.dispatcher
  import spray.json._

  val gymServer: GymServer
  val http = Http(context.system)
  val timeout = 5 seconds

  implicit val materializer: ActorMaterializer = ActorMaterializer(ActorMaterializerSettings(context.system))

  override def preStart() = {
    self ! Initialize
  }

  protected def sendAction(action: Int): Future[StepResponse] = http
    .singleRequest(HttpRequest(uri = gymServer.actEndpoint(action)))
    .flatMap(r => r.entity.toStrict(timeout))
    .map(_.data.decodeString("UTF-8").parseJson.convertTo[StepResponse])

  protected def initialize(): Future[StepResponse] = http
    .singleRequest(HttpRequest(uri = gymServer.resetEndpoint()))
    .flatMap(r => r.entity.toStrict(timeout))
    .map(_.data.decodeString("UTF-8").parseJson.convertTo[InitResponse])
    .map(initResponse => initResponse.toStepResponse)
} 

//设置package包名称以及导入依赖的类
package kneelnrise.warp10scala.services

import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
import akka.stream.Materializer
import akka.stream.scaladsl.{Flow, Source}
import akka.util.ByteString
import kneelnrise.warp10scala.constants.CharsetConstants
import kneelnrise.warp10scala.model.{GTS, Warp10Configuration}

import scala.concurrent.Future
import scala.util.Success

object Warp10CommonTest {
  val emptyPoolClientFlow: Warp10CommonClient.PoolClientFlow = Flow[(HttpRequest, String)].map(_ => (Success(HttpResponse()), ""))
  val warp10Configuration: Warp10Configuration = Warp10Configuration(
    baseUrl = "http://localhost:8080",
    readToken = "rtoken",
    writeToken = "wtoken",
    version = Warp10Configuration.ApiVersion.ZERO
  )

  implicit class SourceByteString(source: Source[ByteString, _]) {
    def toFutureString(implicit materializer: Materializer): Future[String] =
      source.runFold("") { case (acc, current) => acc + current.decodeString(CharsetConstants.`UTF-8`) }
  }

  implicit class SourceGTS(source: Source[GTS, _]) {
    def toFutureSeq(implicit materializer: Materializer): Future[Seq[GTS]] =
      source.runFold(Seq[GTS]()) { case (acc, current) => acc :+ current }
  }

  implicit class StringFold(source: Source[String, _]) {
    def toFutureString(implicit materializer: Materializer): Future[String] =
      source.runFold("") { case (acc, current) => acc + current }
  }
} 

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

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
import akka.stream.scaladsl.{Flow, Keep, Sink, Source}
import akka.stream.{ActorMaterializer, OverflowStrategy, QueueOfferResult}
import rest.client.HttpRequestQueue._

import scala.concurrent.{ExecutionContext, Future, Promise}
import scala.util.{Failure, Success, Try}

trait HttpRequestQueue {

  implicit val system: ActorSystem
  implicit val mat: ActorMaterializer
  implicit val ec: ExecutionContext

  val QueueSize: Int

  val poolClientFlow: Flow[(HttpRequest, Promise[HttpResponse]),
                           (Try[HttpResponse], Promise[HttpResponse]),
                           Http.HostConnectionPool]

  private val queue =
    Source
      .queue[(HttpRequest, Promise[HttpResponse])](QueueSize, OverflowStrategy.backpressure)
      .via(poolClientFlow)
      .toMat(Sink.foreach({
        case ((Success(httpResponse), promisedResponse)) => promisedResponse.success(httpResponse)
        case ((Failure(throwable), promisedResponse))    => promisedResponse.failure(throwable)
      }))(Keep.left)
      .run()

  def queueRequest(request: HttpRequest): Future[HttpResponse] = {
    val responsePromise = Promise[HttpResponse]()
    queue
      .offer(request -> responsePromise)
      .map {
        case QueueOfferResult.Enqueued    => responsePromise
        case QueueOfferResult.Dropped     => responsePromise.failure(new RuntimeException(QUEUE_OVERFLOW))
        case QueueOfferResult.QueueClosed => responsePromise.failure(new RuntimeException(QUEUE_CLOSED))
        case QueueOfferResult.Failure(ex) => responsePromise.failure(ex)
      }
      .flatMap(_.future)
  }
}

object HttpRequestQueue {
  private val QUEUE_OVERFLOW: String = "Queue overflowed. Try again later."
  private val QUEUE_CLOSED: String   = "Queue was closed (pool shut down) while running the request. Try again later."
} 

//设置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包名称以及导入依赖的类
package com.flipkart.connekt.busybees.streams.flows.transformers

import java.util

import akka.http.scaladsl.model.HttpRequest
import akka.http.scaladsl.model.headers.RawHeader
import com.flipkart.connekt.busybees.models.SmsRequestTracker
import com.flipkart.connekt.busybees.streams.errors.ConnektStageException
import com.flipkart.connekt.busybees.streams.flows.MapFlowStage
import com.flipkart.connekt.commons.entities.Channel
import com.flipkart.connekt.commons.factories.{ConnektLogger, LogFile, ServiceFactory}
import com.flipkart.connekt.commons.iomodels.MessageStatus.InternalStatus
import com.flipkart.connekt.commons.iomodels.SmsPayloadEnvelope
import com.flipkart.connekt.commons.services.KeyChainManager
import com.flipkart.connekt.commons.utils.StringUtils.{JSONUnMarshallFunctions, _}

import scala.collection.JavaConverters._

class SmsProviderPrepare extends MapFlowStage[SmsPayloadEnvelope, (HttpRequest, SmsRequestTracker)] {

  lazy implicit val stencilService = ServiceFactory.getStencilService

  override val map: (SmsPayloadEnvelope) => List[(HttpRequest, SmsRequestTracker)] = smsPayloadEnvelope => profile("map") {

    try {
      val selectedProvider = smsPayloadEnvelope.provider.last
      val credentials = KeyChainManager.getSimpleCredential(s"sms.${smsPayloadEnvelope.appName.toLowerCase}.$selectedProvider").get

      val tracker = SmsRequestTracker(messageId = smsPayloadEnvelope.messageId,
        clientId = smsPayloadEnvelope.clientId,
        receivers = smsPayloadEnvelope.payload.receivers,
        provider = selectedProvider,
        appName = smsPayloadEnvelope.appName,
        contextId = smsPayloadEnvelope.contextId,
        request = smsPayloadEnvelope,
        meta = smsPayloadEnvelope.meta)

      val providerStencil = stencilService.getStencilsByName(s"ckt-sms-$selectedProvider").find(_.component.equalsIgnoreCase("prepare")).get

      val result = stencilService.materialize(providerStencil, Map("data" -> smsPayloadEnvelope, "credentials" -> credentials, "tracker" -> tracker).getJsonNode)

      val httpRequests = result.asInstanceOf[util.LinkedHashMap[HttpRequest, String]].asScala.map { case (request, updatedTracker) =>
        (request.addHeader(RawHeader("x-message-id", smsPayloadEnvelope.messageId))
          .addHeader(RawHeader("x-context-id", smsPayloadEnvelope.contextId))
          .addHeader(RawHeader("x-client-id", smsPayloadEnvelope.clientId))
          .addHeader(RawHeader("x-stencil-id", smsPayloadEnvelope.stencilId))
          .addHeader(RawHeader("x-app-name", smsPayloadEnvelope.appName))
          , updatedTracker.getObj[SmsRequestTracker])
      }.toList

      httpRequests
    }
    catch {
      case e: Exception =>
        ConnektLogger(LogFile.PROCESSORS).error(s"SMSChannelFormatter error for ${smsPayloadEnvelope.messageId}", e)
        throw ConnektStageException(smsPayloadEnvelope.messageId, smsPayloadEnvelope.clientId, smsPayloadEnvelope.destinations, InternalStatus.StageError, smsPayloadEnvelope.appName, Channel.SMS, smsPayloadEnvelope.contextId, smsPayloadEnvelope.meta, "SMSChannelFormatter::".concat(e.getMessage), e)
    }
  }

}