//设置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 ru.fediq.scrapingkit

import java.util.concurrent.TimeUnit

import akka.actor.ActorSystem
import akka.http.scaladsl.model.{HttpMethods, Uri}
import com.codahale.metrics.Slf4jReporter
import com.codahale.metrics.Slf4jReporter.LoggingLevel
import com.typesafe.config.ConfigFactory
import org.scalatest.FlatSpec
import org.slf4j.LoggerFactory
import ru.fediq.scrapingkit.backend.{InMemoryFifoLinksQueue, InMemoryLinksHistory, NoOpFeedExporter, NoOpPageCache}
import ru.fediq.scrapingkit.model.PageRef
import ru.fediq.scrapingkit.scraper.HtmlCrawlingScraper
import ru.fediq.scrapingkit.util.Metrics

class ScrapingKitReactorTest extends FlatSpec {

  "Reactor" should "crawl something" in {
    val scraperName = "crawl"
    val scrapers = Map(scraperName -> new HtmlCrawlingScraper(scraperName))

    val config = ConfigFactory.load()
    implicit val system = ActorSystem("reactor-test", config)

    val linksQueue = new InMemoryFifoLinksQueue()
    val linksHistory = new InMemoryLinksHistory()
    val pageCache = new NoOpPageCache()
    val exporter = new NoOpFeedExporter()

    val reactor = new ScrapingKitReactor(linksQueue, linksHistory, pageCache, exporter, scrapers)
    linksQueue.enqueue(PageRef(Uri("http://quotes.toscrape.com/"), HttpMethods.GET, scraperName))

    Slf4jReporter
      .forRegistry(Metrics.metricRegistry)
      .withLoggingLevel(LoggingLevel.INFO)
      .outputTo(LoggerFactory.getLogger("METRICS"))
      .convertDurationsTo(TimeUnit.MILLISECONDS)
      .convertRatesTo(TimeUnit.SECONDS)
      .build()
      .start(10, TimeUnit.SECONDS)

    Thread.sleep(10000)

    reactor.close()
  }
} 

//设置package包名称以及导入依赖的类
package ru.fediq.scrapingkit.model

import akka.http.scaladsl.model.{HttpMethod, HttpMethods, Uri}

case class PageRef(
  uri: Uri,
  method: HttpMethod,
  scraperName: String,
  depth: Double = 1.0,
  context: Map[String, String] = Map.empty,
  redirectsChain: List[Uri] = Nil
) {
  def lastUri = redirectsChain.headOption.getOrElse(uri)

  def redirectSteps = if (redirectsChain.isEmpty) Nil else uri :: redirectsChain.tail

  def chain(nextLocation: Uri) = copy(
    method = HttpMethods.GET,
    redirectsChain = nextLocation :: redirectsChain
  )
} 

//设置package包名称以及导入依赖的类
package ru.fediq.scrapingkit.scraper

import akka.http.scaladsl.model.{HttpMethod, HttpMethods, Uri}
import ru.fediq.scrapingkit.util.Implicits._
import spray.json._

sealed trait Scraped

case class DownloadRequest(
  uri: Uri,
  scraperName: String,
  context: Map[String, String] = Map.empty,
  method: HttpMethod = HttpMethods.GET,
  depthInc: Double = 1.0
) extends Scraped

trait ScrapedEntity extends Scraped {
  def dump: String
}

case class MapScrapedEntity(map: Map[String, Any]) extends ScrapedEntity with DefaultJsonProtocol {
  override def dump = map.mapToJson.compactPrint
}

abstract class JsonScrapedEntity[T](format: JsonFormat[T]) extends ScrapedEntity {
  self: T =>
  override def dump = {
    self.asInstanceOf[T].toJson(format).compactPrint
  }
}

trait PrintableScrapedEntity extends ScrapedEntity {
  override def dump = toString
} 

//设置package包名称以及导入依赖的类
package ru.fediq.scrapingkit.backend

import akka.http.scaladsl.model.{HttpMethods, Uri}
import ru.fediq.scrapingkit.model.PageRef

object RmqConstants {
  val connectorActorName = "links-queue"
  val channelActorName = "rmqchannel"
  val channelActorPath = "/user/links-queue/rmqchannel"

  case class SerializedPageRef(
    uri: String,
    method: String,
    scraper: String,
    depth: Double,
    context: Map[String, String]
  ) {
    def deserialize(): PageRef = PageRef(
      Uri(uri),
      HttpMethods.getForKey(method).getOrElse(HttpMethods.GET),
      scraper,
      depth,
      context
    )
  }

} 

//设置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 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 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 jp.co.dzl.example.akka.api.service

import akka.actor.ActorSystem
import akka.http.scaladsl.model.headers.RawHeader
import akka.http.scaladsl.model.{ HttpMethods, HttpRequest, HttpResponse }
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{ Flow, Source }
import akka.stream.testkit.scaladsl.TestSink
import org.scalamock.scalatest.MockFactory
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.{ BeforeAndAfterAll, FlatSpec, Matchers }

import scala.concurrent.Await
import scala.concurrent.duration.Duration

class GitHubSpec extends FlatSpec with Matchers with ScalaFutures with BeforeAndAfterAll with MockFactory {
  implicit val system = ActorSystem("github-spec")
  implicit val executor = system.dispatcher
  implicit val materializer = ActorMaterializer()

  override protected def afterAll: Unit = {
    Await.result(system.terminate(), Duration.Inf)
  }

  "#from" should "merge original headers to github request" in {
    val github = new GitHubImpl("127.0.0.1", 8000, 5, mock[HttpClient])
    val request = HttpRequest(HttpMethods.GET, "/")
      .addHeader(RawHeader("host", "dummy"))
      .addHeader(RawHeader("timeout-access", "dummy"))

    val result = Source.single(HttpRequest(HttpMethods.GET, "/v1/github/users/xxxxxx"))
      .via(github.from(request))
      .runWith(TestSink.probe[HttpRequest])
      .request(1)
      .expectNext()

    result.headers.filter(_.lowercaseName() == "host") shouldBe empty
    result.headers.filter(_.lowercaseName() == "timeout-access") shouldBe empty
    result.headers.filter(_.lowercaseName() == "x-forwarded-host") shouldNot be(empty)
  }

  "#send" should "connect using http client" in {
    val httpResponse = HttpResponse()
    val httpClient = mock[HttpClient]
    (httpClient.connectionHttps _).expects(*, *, *).returning(Flow[HttpRequest].map(_ => httpResponse))

    val github = new GitHubImpl("127.0.0.1", 8000, 5, httpClient)
    val result = Source.single(HttpRequest(HttpMethods.GET, "/"))
      .via(github.send)
      .runWith(TestSink.probe[HttpResponse])
      .request(1)
      .expectNext()

    result shouldBe httpResponse
  }
} 

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

import akka.http.scaladsl.model.headers._
import akka.http.scaladsl.model.{HttpMethods, HttpResponse, StatusCodes}
import akka.http.scaladsl.server.Directive
import akka.http.scaladsl.server.Directives._

object CorsSupport {

  private val corsHeaders = List(RawHeader("Access-Control-Allow-Origin", "http://localhost:3000"),
    RawHeader("Access-Control-Allow-Methods", "GET, POST, PUT, OPTIONS, DELETE"),
    RawHeader("Access-Control-Allow-Headers", "Origin, X-Requested-With, Content-Type, Accept, Authorization"),
    RawHeader("Access-Control-Allow-Credentials", "true")
  )

  def cors(enabled: Boolean) = {
    import akka.http.scaladsl.server.StandardRoute._
    import akka.http.scaladsl.server.directives.BasicDirectives._

    if (enabled) {
      extractRequest.flatMap[Unit] { request =>
        request.method match {
          case HttpMethods.OPTIONS =>
            complete(HttpResponse(StatusCodes.OK, corsHeaders))
          case _ =>
            respondWithHeaders(corsHeaders)
        }
      }
    } else {
      Directive.Empty
    }
  }

} 

//设置package包名称以及导入依赖的类
package org.eigengo.rsa.ingest.v100

import java.util.UUID

import akka.actor.{Actor, OneForOneStrategy, Props, SupervisorStrategy}
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.{HttpMethods, HttpRequest, Uri}
import akka.stream.ActorMaterializer
import cakesolutions.kafka.{KafkaProducer, KafkaProducerRecord, KafkaSerializer}
import com.google.protobuf.ByteString
import com.typesafe.config.Config
import org.apache.kafka.common.serialization.StringSerializer
import org.eigengo.rsa.Envelope

object SimplifiedTweetProcessorActor {

  def props(config: Config): Props = {
    val producerConf = KafkaProducer.Conf(
      config.getConfig("tweet-image-producer"),
      new StringSerializer,
      KafkaSerializer[Envelope](_.toByteArray)
    )
    Props(classOf[SimplifiedTweetProcessorActor], producerConf)
  }
}

class SimplifiedTweetProcessorActor(producerConf: KafkaProducer.Conf[String, Envelope]) extends Actor {
  private[this] val producer = KafkaProducer(conf = producerConf)
  implicit val _ = ActorMaterializer()

  import scala.concurrent.duration._
  override def supervisorStrategy: SupervisorStrategy = OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 10.seconds) {
    case _ ? SupervisorStrategy.Restart
  }

  override def receive: Receive = {
    case TweetImage(handle, content) ?
      producer.send(KafkaProducerRecord("tweet-image", handle,
        Envelope(version = 100,
          handle = handle,
          ingestionTimestamp = System.nanoTime(),
          processingTimestamp = System.nanoTime(),
          messageId = UUID.randomUUID().toString,
          correlationId = UUID.randomUUID().toString,
          payload = content)))
    case SimplifiedTweet(handle, mediaUrls) ?
      mediaUrls.foreach { mediaUrl ?
        import context.dispatcher
        val request = HttpRequest(method = HttpMethods.GET, uri = Uri(mediaUrl))
        val timeout = 1000.millis
        Http(context.system).singleRequest(request).flatMap(_.entity.toStrict(timeout)).foreach { entity ?
          self ! TweetImage(handle, ByteString.copyFrom(entity.data.toArray))
        }
      }
  }

} 

//设置package包名称以及导入依赖的类
package performanceanalysis.logreceiver.alert

import scala.concurrent.Future
import akka.actor.{Actor, ActorLogging, Props}
import akka.http.scaladsl.{Http, HttpExt}
import akka.http.scaladsl.model.{HttpMethods, HttpRequest, HttpResponse}
import akka.stream.ActorMaterializer
import performanceanalysis.server.messages.AlertMessages.AlertRuleViolated

object AlertActionActor {

  def props(): Props = Props.apply(new AlertActionActor)
}

class AlertActionActor extends Actor with ActorLogging {

  implicit val materializer = ActorMaterializer()

  lazy val http: HttpExt = Http(context.system)

  def receive: Receive = {
    case AlertRuleViolated(endpoint, message) => alert(endpoint, message)
  }

  def alert(endpoint:String, message:String): Unit = {
    //validate endpoint?
    log.info(s"sending alert to $endpoint")
    val req: HttpRequest = HttpRequest(method = HttpMethods.POST, uri = endpoint)
    val resp: Future[HttpResponse] = http.singleRequest(req)
    //check http status of response
    //possibly retry
  }
} 

//设置package包名称以及导入依赖的类
package performanceanalysis.logreceiver.alert

import akka.actor.{ActorSystem, Props}
import akka.event.LoggingAdapter
import akka.http.scaladsl.model.{HttpMethods, HttpRequest, HttpResponse}
import akka.http.scaladsl.settings.ConnectionPoolSettings
import akka.http.scaladsl.{HttpExt, HttpsConnectionContext}
import akka.stream.Materializer
import akka.testkit.TestProbe
import com.typesafe.config.ConfigFactory
import performanceanalysis.base.ActorSpecBase
import performanceanalysis.server.messages.AlertMessages.AlertRuleViolated

import scala.concurrent.Future

class AlertActionActorSpec(testSystem: ActorSystem) extends ActorSpecBase(testSystem) {
  def this() = this(ActorSystem("AlertActionActorSpec"))

  "AlertActionActor" must {

    val endpoint = "http://outlaw.net"
    val req: HttpRequest = HttpRequest(method = HttpMethods.POST, uri = endpoint)

    val alertActionActor = system.actorOf(Props(new AlertActionActor {
      override lazy val http: HttpExt = new HttpExt(ConfigFactory.load()) {
        override def singleRequest(request: HttpRequest,
                                   connectionContext: HttpsConnectionContext,
                                   settings: ConnectionPoolSettings,
                                   log: LoggingAdapter)
                                  (implicit fm: Materializer): Future[HttpResponse] = {
          if (request != req) {
            throw new UnsupportedOperationException
          } else {
            super.singleRequest(request, connectionContext, settings, log)
          }
        }
      }
    }))

    "send out an alert to a given endpoint when it receives an AlertingRuleViolated message " in {
      val testProbe = TestProbe("testProbe")

      testProbe.send(alertActionActor, AlertRuleViolated(endpoint, "One of your loglines violated an alert rule"))

      testProbe.expectNoMsg()
    }
  }
} 

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

import akka.http.scaladsl.model.headers.RawHeader
import akka.http.scaladsl.model.{HttpMethods, HttpRequest}
import org.scalatest.FunSuite

class QSSignerTest extends FunSuite {

  test("Test getHeadAuthorization") {
    val url =
      "https://qingstor.com/?acl&upload_id=fde133b5f6d932cd9c79bac3c7318da1&part_number=0&other=abc"
    val headers = List(RawHeader("X-QS-Test-2", "Test 2"),
                       RawHeader("X-QS-Test-1", "Test 1"),
                       RawHeader("Date", "Mon, 01 Jan 0001 00:00:00 GMT"))
    val request = HttpRequest()
      .withUri(url)
      .withMethod(HttpMethods.GET)
      .withHeaders(headers)
    val signature = "QS ENV_ACCESS_KEY_ID:bvglZF9iMOv1RaCTxPYWxexmt1UN2m5WKngYnhDEp2c="
    val authorization = QSSigner.getHeadAuthorization(request, "ENV_ACCESS_KEY_ID", "ENV_SECRET_ACCESS_KEY")
    assert(authorization == signature)
  }

  test("Test getHeadAuthorization chinese") {
    val url = "https://zone.qingstor.com/bucket-name/??"
    val headers = List(RawHeader("Date", "Mon, 01 Jan 0001 00:00:00 GMT"))
    val request = HttpRequest()
      .withUri(new java.net.URI(url).toASCIIString)
      .withMethod(HttpMethods.GET)
      .withHeaders(headers)
    val authorization = QSSigner.getHeadAuthorization(request, "ENV_ACCESS_KEY_ID", "ENV_SECRET_ACCESS_KEY")
    val signature = "QS ENV_ACCESS_KEY_ID:XsTXX50kzqBf92zLG1aIUIJmZ0hqIHoaHgkumwnV3fs="
    assert(authorization == signature)
  }

  test("Test getQueryAuthorization") {
    val url =
      "https://qingstor.com/?acl&upload_id=fde133b5f6d932cd9c79bac3c7318da1&part_number=0&other=abc"
    val headers = List(RawHeader("X-QS-Test-2", "Test 2"),
      RawHeader("X-QS-Test-1", "Test 1"),
      RawHeader("Date", "Mon, 01 Jan 0001 00:00:00 GMT"))
    val request = HttpRequest()
      .withUri(url)
      .withMethod(HttpMethods.GET)
      .withHeaders(headers)
    val auth = QSSigner.getQueryAuthorization(request, "ENV_ACCESS_KEY_ID", "ENV_SECRET_ACCESS_KEY", -62135596800L)
    val signature = Map(
      "access_key_id" -> "ENV_ACCESS_KEY_ID",
      "expires" -> "-62135596800",
      "signature" -> "gTdB/cmD6rjv8CbFRDfFbHc64q442rYNAp99Hm7fBl4="
    )
    assert(auth == signature)
  }
} 

//设置package包名称以及导入依赖的类
package io.cronit.utils

import akka.http.scaladsl.model.{HttpMethod, HttpMethods}

object HttpMethodUtils {

  implicit class HttpMethodConverter(method: String) {
    def toHttpMethod: HttpMethod = {
      val methodToUpper = method.toUpperCase
      methodToUpper match {
        case "GET" => HttpMethods.GET
        case "POST" => HttpMethods.POST
        case "PUT" => HttpMethods.PUT
        case "DELETE" => HttpMethods.DELETE
        case "PATCH" => HttpMethods.PATCH
      }
    }
  }

} 

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

import akka.http.scaladsl.model.{FormData, HttpMethods, HttpRequest}
import com.github.chaabaj.openid.HttpClient
import com.github.chaabaj.openid.exceptions.{OAuthException, WebServiceException}
import spray.json.{JsValue, JsonFormat}

import scala.concurrent.{ExecutionContext, Future}

trait SupportsIssuingAccessToken { self: OAuthClient =>

  protected def  accessTokenUrl: String
  def httpClient: HttpClient
  def issueOAuthToken(request: AccessTokenRequest)
    (implicit exc: ExecutionContext, jfs: JsonFormat[AccessTokenSuccess], jfe: JsonFormat[Provider#AccessTokenError]):
  Future[AccessTokenSuccess] = {
    val httpRequest = HttpRequest(
      HttpMethods.POST,
      uri = accessTokenUrl,
      entity = FormData(
        "client_id" -> request.clientId,
        "client_secret" -> request.clientSecret,
        "code" -> request.code,
        "redirect_uri" -> request.redirectUri,
        "grant_type" -> "authorization_code"
      ).toEntity
    )
    httpClient.request(httpRequest)
      .map(_.convertTo[AccessTokenSuccess])
      .recover {
        case WebServiceException(status, jsonError: JsValue) =>
          throw OAuthException(status, jsonError.convertTo[Provider#AccessTokenError])
        case t: Throwable =>
          throw t
      }
  }
} 

//设置package包名称以及导入依赖的类
package io.tardieu.netwemo.connectors

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.Uri.Query
import akka.http.scaladsl.model.{HttpMethods, HttpRequest, Uri}
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.stream.ActorMaterializer
import com.typesafe.config.ConfigFactory
import org.slf4j.LoggerFactory

import scala.util.{Failure, Success}


class WemoConnector(implicit val system: ActorSystem) {

  private[this] val logger = LoggerFactory.getLogger(getClass)

  private[this] implicit val materializer = ActorMaterializer()
  private[this] implicit val executionContext = system.dispatcher

  private[this] val conf = ConfigFactory.load.getConfig("wemo")
  private[this] val wemoHost = conf.getString("host")
  private[this] val wemoPort = conf.getInt("port")

  def switchOn(device: String): Unit = switchState("on", device)
  def switchOff(device: String): Unit = switchState("off", device)
  def toggle(device: String): Unit = switchState("toggle", device)

  private def switchState(state: String, device: String): Unit = {
    val request = HttpRequest(
      method = HttpMethods.POST,
      uri = Uri(s"http://$wemoHost:$wemoPort/api/device/$device").withQuery(Query("state" -> state))
    )

    Http().singleRequest(request).onComplete {
      case Success(response) =>
        val message = Unmarshal(response.entity).to[String]
        if (response.status.isFailure()) {
          message.foreach(logger.error(s"HTTP error {}: {}", response.status.asInstanceOf[Any], _))
        }
        else {
          logger.debug(
            "Switch {} order for {} received with status {}",
            state, device, response.status)
          logger.trace(
            "Switch {} order for {} received with status {}: {}",
            state, device, response.status, message)
        }
      case Failure(e) =>
        logger.error("Error contacting the wemo server", e)
    }
  }

} 

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

import java.util.UUID

import akka.actor.ActorSystem
import akka.http.scaladsl.model.{HttpEntity, HttpMethods, HttpRequest}
import akka.stream.ActorMaterializer
import akka.stream.testkit.scaladsl.TestSink
import akka.testkit.TestKit
import org.scalatest.{Matchers, WordSpecLike}

import scala.concurrent.Future


class StreamsSpec extends TestKit(ActorSystem("StreamsSpec"))
                          with WordSpecLike with Matchers {

  implicit val mat = ActorMaterializer()

  "request factory" should {
    val path = s"/${UUID.randomUUID.toString.take(5)}"
    val expected = HttpRequest(uri = uripath(path))

    "provide empty GET" when {
      val expectedEmpty = expected.withEntity(HttpEntity.Empty)

      "GET factory called" in {
        Streams.get(path)
        .runWith(TestSink.probe[HttpRequest])
        .request(1)
        .expectNext(expectedEmpty)
        .expectComplete()
      }

      "request defaults to GET" in {
        Streams.request(path)(Future.successful)
        .runWith(TestSink.probe[HttpRequest])
        .request(1)
        .expectNext(expectedEmpty)
        .expectComplete()
      }
    }

    "provide modified request" when {
      "request set to POST" in {
        Streams.request(path)(r => Future.successful(r.withMethod(HttpMethods.POST)))
        .runWith(TestSink.probe[HttpRequest])
        .request(1)
        .expectNext(expected.withMethod(HttpMethods.POST))
        .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 service

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.stream.ActorMaterializer

import scala.concurrent.{ExecutionContextExecutor, Future}
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
import akka.http.scaladsl.model.{HttpMethods, HttpRequest}
import akka.http.scaladsl.unmarshalling.Unmarshal
import com.typesafe.config.ConfigFactory
import spray.json.DefaultJsonProtocol

case class GoogleTokenInfo(aud: String, sub: String, email: String)

sealed trait TokenType
case object IdToken     extends TokenType
case object AccessToken extends TokenType

class GoogleOauthService()(implicit system: ActorSystem, executor: ExecutionContextExecutor, materializer: ActorMaterializer)
    extends DefaultJsonProtocol {

  implicit val googleTokenInfoFormat = jsonFormat3(GoogleTokenInfo)
  val config                         = ConfigFactory.load()

  val fetchAndValidateTokenInfo: (String, TokenType) => Future[GoogleTokenInfo] = (tokenValue, tokenType) => {
    val tokenQuery = tokenType match {
      case IdToken     => s"id_token=$tokenValue"
      case AccessToken => s"access_token=$tokenValue"
    }

    for {
      response <- Http().singleRequest(
        HttpRequest(
          method = HttpMethods.GET,
          uri = s"https://www.googleapis.com/oauth2/v3/tokeninfo?$tokenQuery"
        ))
      tokenInfo <- Unmarshal(response.entity).to[GoogleTokenInfo]
      validatedTokenInfo <- if (tokenInfo.aud == config.getString("googleClientId")) Future.successful(tokenInfo)
      else Future.failed(new RuntimeException("Invalid sub"))
    } yield validatedTokenInfo
  }
}