//设置package包名称以及导入依赖的类
import akka.actor.ActorSystem
import akka.io.IO
import io.prediction.controller.Engine
import io.prediction.data.storage.EngineManifest
import io.prediction.workflow.{CreateServer, ServerConfig, WorkflowUtils}
import spray.can.Http
import spray.can.server.ServerSettings

object ServerApp extends App {

  val port = sys.env.getOrElse("PORT", "8000").toInt
  val eventServerIp = sys.env.getOrElse("EVENT_SERVER_IP", "localhost")
  val eventServerPort = sys.env.getOrElse("EVENT_SERVER_PORT", "7070").toInt
  val maybeAccessKey = sys.env.get("ACCESS_KEY")

  val maybeLatestEngineInstance = CreateServer.engineInstances.getLatestCompleted(EngineConfig.engineId, EngineConfig.engineVersion, EngineConfig.engineVariantId)

  maybeLatestEngineInstance.map { engineInstance =>
    // the spark config needs to be set in the engineInstance
    engineInstance.copy(sparkConf = engineInstance.sparkConf.updated("spark.master", "local"))
  }.fold {
    println("Could not get latest completed engine instance")
  } { engineInstance =>

    val sc = ServerConfig(
      engineInstanceId = engineInstance.id,
      engineId = Some(engineInstance.engineId),
      engineVersion = Some(engineInstance.engineVersion),
      engineVariant = EngineConfig.engineVariant,
      port = port,
      eventServerIp = eventServerIp,
      eventServerPort = eventServerPort,
      accessKey = maybeAccessKey
    )

    val (engineLanguage, engineFactory) = WorkflowUtils.getEngine(engineInstance.engineFactory, getClass.getClassLoader)
    val engine = engineFactory()

    // WTF: The list of files must be at least 2 long due to https://github.com/PredictionIO/PredictionIO/blob/v0.9.6/core/src/main/twirl/io/prediction/workflow/index.scala.html#L56
    val manifest = EngineManifest(engineInstance.engineId, engineInstance.engineVersion, "default", None, Seq("WHAT", "THE"), engineInstance.engineFactory)

    val actor = CreateServer.createServerActorWithEngine(
      sc,
      engineInstance,
      engine.asInstanceOf[Engine[_, _, _, _, _, _]],
      engineLanguage,
      manifest
    )

    val actorSystem = ActorSystem("pio-server")

    val settings = ServerSettings(actorSystem)

    IO(Http)(actorSystem) ! Http.Bind(listener = actor, interface = sc.ip, port = sc.port)

    actorSystem.awaitTermination()
  }

} 

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

import akka.actor.{ActorSystem, Props}
import akka.io.IO
import akka.pattern._
import akka.util.Timeout

import scala.concurrent.Await
import scala.concurrent.duration._

import spray.can.Http
import spray.can.server.ServerSettings
import spray.routing._
import Directives._

class GoogleBooksApiStub(val route: Route) {
  implicit val system = ActorSystem("google-books-api")
  implicit val timeout: Timeout = 3 seconds 

  val settings = ServerSettings(system).copy(sslEncryption = false)
  val handler = system.actorOf(Props(new GoogleBooksRestService(route)), name = "handler")

  def start(port: Int) = 
    Await.ready(IO(Http) ? Http.Bind(handler, 
      interface = "localhost", port = port, settings = Some(settings)), timeout.duration)
      
  def stop() = {
    IO(Http) ? Http.CloseAll
    system.stop(handler)
  }
}

sealed class GoogleBooksRestService(val route: Route) extends HttpServiceActor {
  def receive = runRoute {
    route
  }
} 

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

import akka.actor.{Props, ActorSystem, Actor}
import akka.io.IO
import spray.can.Http
import spray.routing.HttpService
import spray.http.MediaTypes._

trait MyService extends HttpService {
  val myRoute = {
    path("something") {
      respondWithMediaType(`text/plain`) {
        complete("okay")
      }
    }
    path("hello") {
      respondWithMediaType(`text/plain`) {
        complete("kitty")
      }
    }
  }
}

class MyServiceActor extends Actor with MyService {
  def actorRefFactory = context
  def receive = runRoute(myRoute)
}

object server extends App {
  try {
    implicit val system = ActorSystem("my-system")
    val handler = system.actorOf(Props[MyServiceActor], name = "my-service")

    IO(Http) ! Http.Bind(handler, interface = "localhost", port = 8080)
  } catch {
    case e: Throwable => println("error: ", e)
  }
} 

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

import boxOffice.RestInterface
import akka.actor.{Props, ActorSystem}
import com.typesafe.config.ConfigFactory
import spray.can.Http
import spray.can.Http.Bind

object BackendRemoteDeployMain extends App {
  val config = ConfigFactory.load("backend")
  val system = ActorSystem("backend", config)
}

object FrontendRemoteDeployMain extends App {
  val config = ConfigFactory.load("frontend-remote-deploy")
  val host = config.getString("http.host")
  val port = config.getInt("http.port")
  val system = ActorSystem("frontend", config)

  val restInterface = system.actorOf(Props[RestInterface],
    "restInterface")
  Http(system).manager ! Bind(listener = restInterface,
    interface = host, port = port)
} 

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

import akka.actor.{ActorSystem, Props}
import akka.io.IO
import com.sretsnom.mangareader.server.ServiceConfig.HttpConfig._
import com.typesafe.scalalogging.LazyLogging
import spray.can.Http

object ServerApp extends App with LazyLogging {
    implicit val system = ActorSystem("mr-server")
    val httpServer = system.actorOf(Props[HttpServerActor], "httpserver")

    logger.info(banner)

    IO(Http) ! Http.Bind(httpServer, interface, port)

    private def banner =
        """
       =============================================================

                        `-:-                  .:-.
                  `-+shmNNNo                  :NNNmdy+:`
                -hmNNNNNNNNo                  :NNNNNNNNNd/
                `hNNNNNNNNNo                  :NNNNNNNNNm.
                 `omNNNNNNNs                  /NNNNNNNNy.
                   `+hNNNNNN+:/+oo+////+ooo+//mNNNNNdo.
                      .sNNNNNNmo.        `/dNNNNNNh:
                   `+hNNNNNN+:/+oo+////+ooo+//mNNNNNdo.
                 `omNNNNNNNs                  /NNNNNNNNy.
                `hNNNNNNNNNo                  :NNNNNNNNNm.
                -hmNNNNNNNNo                  :NNNNNNNNNd/
                  `-+sdmNNNo                  :NNNmdy+:`
                        `-:-                  .:-.
       =============================================================
                         IDM SELF SERVICE TOOLING
        """.stripMargin
} 

//设置package包名称以及导入依赖的类
package eu.shiftforward.apso.spray

import akka.actor.ActorSystem
import akka.io.IO
import spray.can.Http
import spray.http.{ HttpHeader, HttpRequest, RemoteAddress, Uri }
import spray.http.HttpHeaders.{ `Remote-Address`, `X-Forwarded-For` }
import spray.routing.{ Directive1, RequestContext, Route }


trait ProxySupport extends ClientIPDirectives {

  private def sending(f: RequestContext => HttpRequest)(implicit system: ActorSystem): Route = {
    val transport = IO(Http)(system)
    ctx => transport.tell(f(ctx), ctx.responder)
  }

  private def getHeaders(ip: Option[RemoteAddress], headers: List[HttpHeader] = Nil) = {
    // filter `Host` header
    val hs = headers.filterNot(header => header.is("host"))
    // add `X-Forwarded-For` header
    ip.fold(hs)(addForwardedFor(_, hs))
  }

  private def addForwardedFor(ip: RemoteAddress, headers: List[HttpHeader]): List[HttpHeader] = {
    headers match {
      case Nil =>
        // No `X-Forwarded-For` found in headers, so just add the new one
        `X-Forwarded-For`(ip) :: Nil

      case `X-Forwarded-For`(ips) :: tail =>
        `X-Forwarded-For`(ips :+ ip) :: tail

      case notForwardedFor :: tail =>
        notForwardedFor :: addForwardedFor(ip, tail)
    }
  }

  private val optionalRemoteAddress: Directive1[Option[RemoteAddress]] =
    headerValuePF { case `Remote-Address`(address) => Some(address) } | provide(None)

  def proxyTo(uri: Uri)(implicit system: ActorSystem): Route = {
    optionalRemoteAddress { ip =>
      sending(ctx => ctx.request.copy(
        uri = uri,
        headers = getHeaders(ip, ctx.request.headers)))
    }
  }

  def proxyToUnmatchedPath(uri: Uri)(implicit system: ActorSystem): Route = {
    optionalRemoteAddress { ip =>
      sending { ctx =>
        ctx.request.copy(
          uri = uri.withPath(uri.path.++(ctx.unmatchedPath)).withQuery(ctx.request.uri.query),
          headers = getHeaders(ip, ctx.request.headers))
      }
    }
  }
} 

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

import akka.actor.{Actor, Props, ActorSystem}
import spray.can.Http
import spray.http._
import spray.http.HttpRequest
import spray.http.HttpResponse
import akka.io.IO
import scala.io.Source

class TwitterApi private(system: ActorSystem, port: Int, body: String) {

  private class Service extends Actor {

    def receive: Receive = {
      case _: Http.Connected =>
        sender ! Http.Register(self)
      case HttpRequest(HttpMethods.POST, _, _, _, _) =>
        sender ! ChunkedResponseStart(HttpResponse(StatusCodes.OK))
        sender ! MessageChunk(body = body)
        sender ! ChunkedMessageEnd()
    }
  }

  private val service = system.actorOf(Props(new Service))
  private val io = IO(Http)(system)
  io ! Http.Bind(service, "localhost", port = port)

  def stop(): Unit = {
    io ! Http.Unbind
    system.stop(service)
    system.stop(io)
  }
}

object TwitterApi {

  def apply(port: Int)(implicit system: ActorSystem): TwitterApi = {
    val body = Source.fromInputStream(getClass.getResourceAsStream("/tweet.json")).mkString
    new TwitterApi(system, port, body)
  }

} 

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

import akka.actor.{Actor, ActorLogging, Props}
import spray.can.Http



class WebSocketServer extends Actor with ActorLogging {
  def receive = {
    // when a new connection comes in we register a WebSocketConnection actor as the per connection handler
    case Http.Connected(remoteAddress, localAddress) =>
      val serverConnection = sender()
      val conn = context.actorOf(Props(classOf[WebSocketWorker], serverConnection))
      serverConnection ! Http.Register(conn)
    case PushToChildren(msg: String) =>
      val children = context.children
      println("pushing to all children : " + msg)
      children.foreach(ref => ref ! Push(msg))
  }
} 

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

object WebServer extends App {
  import akka.actor.{ActorSystem, Props}
  import akka.util.Timeout
  import akka.io.IO
  import akka.pattern.ask
  import spray.can.Http
  import scala.concurrent._
  import scala.concurrent.duration._
  slogging.LoggerConfig.factory = slogging.SLF4JLoggerFactory()
  slogging.LoggerConfig.level = slogging.LogLevel.DEBUG

  implicit val system = ActorSystem("server")
  implicit val timeout = Timeout(120.seconds) // TODO(arjun): Add comment noting what this timeout is for

  val server = system.actorOf(Props(classOf[WebServerActor]), "web-server")
  val io = IO(Http)
  // Wait for up to ten seconds to bind to server port, then fail.
  Await.result(io.ask(Http.Bind(server, interface = Settings.bindAddress, port = 8080)), 10.seconds)
} 

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

import akka.actor.{ ActorSystem, Props, _ }
import akka.io.IO
import akka.util.Timeout
import com.weibo.datasys.rest.Configuration
import spray.can.Http
import scala.concurrent.duration._


object Main
    extends Configuration {

  def main(args: Array[String]): Unit = {
    lazy val cmd = new ArgumentConf(args)

    if (cmd.help()) {
      cmd.printHelp()
      sys.exit(0)
    }

    if (cmd.rest_service()) {
      startRestService()
    } else if (cmd.scheduler_service()) {
      startJobSchedulerService()
    }

    def startRestService() = {
      implicit val system = ActorSystem(
        cluster_name,
        config.getConfig("rest-service").withFallback(config)
      )
      implicit val executionContext = system.dispatcher
      implicit val timeout = Timeout(10 seconds)

      val restService = system.actorOf(Props[RestServiceActor], RestServiceActor.Name)
      IO(Http) ! Http.Bind(restService, host, port)
    }

    def startJobSchedulerService() = {
      implicit val system = ActorSystem(
        cluster_name,
        config.getConfig("scheduler-service").withFallback(config)
      )
      system.actorOf(JobSchedulerActor.props(), JobSchedulerActor.Name)
    }
  }
} 

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

import akka.actor.{Actor, ActorLogging}
import spark.jobserver.ChunkEncodingActor.Ok
import spray.can.Http
import spray.http._
import spray.routing.RequestContext

object ChunkEncodingActor {
  // simple case class whose instances we use as send confirmation message for streaming chunks
  case class Ok(remaining: Iterator[_])
}


class ChunkEncodingActor(ctx: RequestContext,
                         chunkSize: Int,
                         byteIterator: Iterator[_]) extends Actor with ActorLogging {
  // we use the successful sending of a chunk as trigger for sending the next chunk
  ctx.responder ! ChunkedResponseStart(
    HttpResponse(entity = HttpEntity(MediaTypes.`application/json`,
      byteIterator.take(chunkSize).map {
        case c: Byte => c
      }.toArray))).withAck(Ok(byteIterator))

  def receive: Receive = {
    case Ok(remaining) =>
      val arr = remaining.take(chunkSize).map {
        case c: Byte => c
      }.toArray
      if (arr.nonEmpty) {
        ctx.responder ! MessageChunk(arr).withAck(Ok(remaining))
      }
      else {
        ctx.responder ! ChunkedMessageEnd
        context.stop(self)
      }
    case ev: Http.ConnectionClosed => {
      log.warning("Stopping response streaming due to {}", ev)
      context.stop(self)
    }
  }
} 

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

import akka.actor.{Actor, ActorRef, Props}
import com.typesafe.config.ConfigFactory
import greendash.dataplayer.Clock.AdjustSpeed
import spray.can.Http
import spray.http.HttpMethods._
import spray.http.{HttpRequest, Uri, _}


class RestService(clock: ActorRef) extends Actor {

    val config = ConfigFactory.load()
    var speedFactor = config.getInt("speed.factor")

    override def receive: Receive = {

        // when a new connection comes in we register ourselves as the connection handler
        case _: Http.Connected => sender ! Http.Register(self)

        case HttpRequest(POST, Uri.Path("/speed"), _, entity: HttpEntity.NonEmpty, _) =>
            try {
                speedFactor = entity.asString.toInt
                sender ! HttpResponse(entity = "OK")
                clock ! AdjustSpeed(speedFactor)
            } catch {
                case e: Throwable =>
                    sender ! HttpResponse(entity = "NOK: " + e.getMessage)
            }

        case HttpRequest(GET, Uri.Path("/speed"), _, entity, _) =>
            sender ! HttpResponse(entity = speedFactor.toString)

        case HttpRequest(GET, Uri.Path("/sensorList"), _, entity: HttpEntity, _) =>
            val target = sender
            clock ! ForwardSensorList(target, entity)

        case HttpRequest(GET, Uri.Path("/state"), _, entity: HttpEntity, _) =>
            val target = sender
            clock ! ForwardState(target, entity)

    }
}

object RestService {
    def props(clock: ActorRef) = Props(new RestService(clock))
}

class ForwardHttpResponse(target: ActorRef, entity: HttpEntity) {
    def forward(answer: String) = target ! HttpResponse(entity = answer)
}

case class ForwardSensorList(target: ActorRef, entity: HttpEntity) extends ForwardHttpResponse(target, entity)
case class ForwardState(target: ActorRef, entity: HttpEntity) extends ForwardHttpResponse(target, entity) 

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

import akka.actor.ActorSystem
import akka.io.IO
import com.typesafe.config.ConfigFactory
import greendash.dataplayer.Clock.Start
import spray.can.Http

object ApplicationMain extends App {
    implicit val system = ActorSystem("DataPlayerSystem")
    val clock = system.actorOf(Clock.props())
    clock ! Start

    val config = ConfigFactory.load()
    var restPort = config.getInt("rest.port")
    val handler = system.actorOf(RestService.props(clock))
    IO(Http) ! Http.Bind(handler, interface = "localhost", port = restPort)

    system.awaitTermination()
} 

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

import akka.actor.{Actor, ActorSystem, Props}
import akka.io.IO
import spray.can.Http
import spray.http.{HttpRequest, HttpResponse}
import spray.http.HttpHeaders.Connection

object Main extends App {
  implicit val system: ActorSystem = ActorSystem()
  val boot = system.actorOf(Props(classOf[Boot]))
}

class Boot extends Actor {
  import context.system

  IO(Http) ! Http.Bind(self, interface = "0.0.0.0", port = 80)

  def receive = {
    case _: Http.Connected => sender ! Http.Register(self)
    case _: HttpRequest => sender ! HttpResponse(headers = List(`Connection`("keep-alive")))
  }
} 

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

import akka.actor._
import akka.io.IO
import akka.pattern.ask
import akka.util.Timeout
import com.typesafe.config.ConfigFactory
import spray.can.Http

import scala.concurrent.duration._

object Main extends App {
  val config = ConfigFactory.load()
  val host = config.getString("http.host")
  val port = config.getInt("http.port")

  implicit val system = ActorSystem("mikeapi-management-service")

  val api = system.actorOf(Props(new RestInterface()), "httpInterface")

  implicit val executionContext = system.dispatcher
  implicit val timeout = Timeout(10 seconds)

  IO(Http).ask(Http.Bind(listener = api, interface = host, port = port))
    .mapTo[Http.Event]
    .map {
      case Http.Bound(address) =>
        println(s"[[MIKE]] - REST interface bound to $address")
      case Http.CommandFailed(cmd) =>
        println("[[MIKE]] - REST interface could not bind to " +
          s"$host:$port, ${cmd.failureMessage}")
        system.shutdown()
    }
}