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

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

import scala.concurrent.ExecutionContextExecutor
import scala.concurrent.duration._

object Boot extends VoteService {

  implicit val system = ActorSystem("user-system")
  implicit val timeout: Timeout = Timeout(10.seconds)
  implicit val executionContext: ExecutionContextExecutor = system.dispatcher
  implicit val materializer = ActorMaterializer()

  def main(args: Array[String]) {
    val port = 5000
    val bindingFuture = Http().bindAndHandle(routes, "0.0.0.0", port)
    println(s"Server online at http://localhost:$port/")
//    bindingFuture
//      .onComplete(e => {
//        println(s"Binding failure, terminating: ${e}")
//        system.terminate()
//      }) // and shutdown when done
  }
} 

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

import akka.actor.ActorSystem
import akka.event.LoggingAdapter
import akka.http.scaladsl.marshalling.ToResponseMarshallable
import akka.http.scaladsl.server.Route
import akka.http.scaladsl.server.Directives._
import akka.stream.Materializer
import com.ferhtaydn.models.{ JsonFormats, PatientInfo, PatientResult }
import com.ferhtaydn.rater.RatePredictor
import akka.http.scaladsl.model.StatusCodes._

import scala.concurrent.ExecutionContextExecutor

trait Server extends JsonFormats {

  implicit val system: ActorSystem

  implicit def executor: ExecutionContextExecutor

  implicit val materializer: Materializer

  val logger: LoggingAdapter

  def predictor: RatePredictor

  protected val routes: Route =
    pathPrefix("cancerater") {
      get {
        pathSingleSlash {
          complete("Welcome to CanceRater")
        }
      } ~
        get {
          path("cm") {
            complete {
              predictor.confusionMatrixString.map[ToResponseMarshallable] {
                case cm ? cm
              }
            }
          }
        } ~
        post {
          path("check") {
            entity(as[PatientInfo]) { patientInfo ?
              complete {
                predictor.predict(patientInfo).map[ToResponseMarshallable] {
                  case Right(score) ? PatientResult(score)
                  case Left(error)  ? BadRequest ? error
                }
              }
            }
          }
        }
    }
} 

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

import akka.actor.ActorSystem
import com.ferhtaydn.models.PatientInfo
import org.apache.spark.ml.feature.StringIndexerModel
import org.apache.spark.ml.tuning.CrossValidatorModel
import org.apache.spark.mllib.linalg.{ Matrix, Vector }
import org.apache.spark.sql.{ Row, SQLContext }

import scala.concurrent.{ ExecutionContextExecutor, Future }

class RatePredictor(system: ActorSystem, sqlContext: SQLContext,
    indexModel: StringIndexerModel, cvModel: CrossValidatorModel,
    confusionMatrix: String) {

  private val decimalFormatter = new java.text.DecimalFormat("##.##")
  private val blockingDispatcher: ExecutionContextExecutor = system.dispatchers.lookup("ml.predictor.dispatcher")

  def confusionMatrixString: Future[String] = {
    Future {
      confusionMatrix
    }(blockingDispatcher)
  }

  def predict(patientInfo: PatientInfo): Future[Either[String, Double]] = {

    Future {

      val df = sqlContext.createDataFrame(Seq(patientInfo.toRecord))
      val indexedJobDF = indexModel.transform(df)

      val result = cvModel
        .transform(indexedJobDF)
        .select("prediction", "probability").map {
          case Row(prediction: Double, probability: Vector) ?
            (probability, prediction)
        }

      result.collect().headOption match {
        case Some((prob, _)) ? Right(decimalFormatter.format(prob(1)).toDouble)
        case None            ? Left(s"No result can be predicted for the patient")
      }

    }(blockingDispatcher)
  }

} 

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

import akka.actor.ActorSystem
import akka.http.scaladsl._
import akka.stream.ActorMaterializer
import dummy_authenticator.config.Configuration
import dummy_authenticator.rest.Routes

import scala.concurrent.ExecutionContextExecutor
import scala.io.StdIn

object WebServer extends Routes {
  val config: Configuration                 = Configuration.get
  implicit val system                       = ActorSystem(config.app.name)
  implicit val materializer                 = ActorMaterializer()
  implicit val ec: ExecutionContextExecutor = system.dispatcher

  def main(args: Array[String]) {
    val interface = config.server.interface
    val port      = config.server.dummy_port

    // The Magic Words
    val bindingFuture = Http().bindAndHandle(routes, interface, port)

    // Console "Controls"
    println(s"dummy_authenticator online at http://$interface:$port/\nPress RETURN to stop...")
    StdIn.readLine() // let it run until user presses return

    // Closing Shop
    bindingFuture
      .flatMap(_.unbind()) // trigger unbinding from the port
      .onComplete(_ => system.terminate()) // and shutdown when done
  }
} 

//设置package包名称以及导入依赖的类
package co.horn.alkes.server

import akka.actor.ActorSystem
import akka.http.scaladsl._
import akka.stream.ActorMaterializer
import co.horn.alkes.auth._
import co.horn.alkes.config.Configuration
import co.horn.alkes.dao.DataHandler
import co.horn.alkes.dao.implementations.riak.RiakDataHandler
import co.horn.alkes.log.Logger
import co.horn.alkes.rest.Routes

import scala.concurrent.ExecutionContextExecutor
import scala.io.StdIn

object WebServer extends Routes {
  implicit val system                       = ActorSystem("alkes")
  implicit val materializer                 = ActorMaterializer()
  implicit val ec: ExecutionContextExecutor = system.dispatcher
  val config: Configuration                 = Configuration.get
  val dao: DataHandler                      = new RiakDataHandler(config)
  val auth: Authority                       = new WasatAuthority(config)
  val log: Logger                           = config.log.server

  def main(args: Array[String]) {
    val interface = config.server.interface
    val port      = config.server.port

    // The Magic Words
    val bindingFuture = Http().bindAndHandle(routes, interface, port)

    // Console "Controls"
    println(s"alkes online at http://$interface:$port/\nPress RETURN to stop...")
    StdIn.readLine() // let it run until user presses return

    // Closing Shop
    bindingFuture
      .flatMap(_.unbind()) // trigger unbinding from the port
      .onComplete(_ => system.terminate()) // and shutdown when done
  }
} 

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

import akka.actor.ActorSystem
import akka.http.scaladsl.model._
import akka.stream.ActorMaterializer
import edu.goldlok.minio_scala.auth.MioKeys

import scala.concurrent.{ExecutionContextExecutor, Future}


  def deleteBucket(bucket: String)
                  (implicit system: ActorSystem,
                   mater: ActorMaterializer,
                   dispatcher: ExecutionContextExecutor): Future[Boolean] = {
    for {
      result <- deleteBucketRaw(bucket)
    } yield {
      if (result.isSuccess) true
      else {
        if (result.ex.isDefined) throw result.ex.get
        else throw new Throwable(s"minio server response status not ok! status: ${result.status}")
      }
    }
  }
} 

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

import akka.actor.ActorSystem
import akka.http.scaladsl.model._
import akka.stream.ActorMaterializer
import edu.goldlok.minio_scala.auth.MioKeys

import scala.concurrent.{ExecutionContextExecutor, Future}


  def makeBucket(bucket: String)
                (implicit system: ActorSystem,
                mater: ActorMaterializer,
                dispatcher: ExecutionContextExecutor): Future[Boolean] = {
    val response = this.makeBucketRaw(bucket)
    for {
      result <- response
    } yield {
      if (result.isSuccess) true
      else {
        if (result.ex.isDefined) throw result.ex.get
        else throw new Throwable(s"minio server response status not ok! status: ${result.status}")
      }
    }
  }
} 

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

import akka.actor.{ActorSystem, Terminated}
import akka.stream.ActorMaterializer
import edu.goldlok.minio_scala.auth.MioKeys
import edu.goldlok.minio_scala.mio.MioClient
import io.minio.MinioClient

import scala.concurrent.duration._
import scala.concurrent.{ExecutionContextExecutor, Future}


object MioSystem {
  implicit val mioSystem = ActorSystem("mio_system")
  implicit val mater =  ActorMaterializer()
  implicit val dispatcher: ExecutionContextExecutor = mioSystem.dispatcher
  implicit val keys = MioKeys("12345678", "12345678")
  implicit val bucket = "guangzhou"
  implicit val notfoundBucket = "guangzhou1"
  implicit val timeout: FiniteDuration = 180.seconds
  private val host = "192.168.2.245"
  private val port = 9001
  implicit val mc = MioClient(host, 9001, keys, blockSize = 5 * 1024 * 1024)
  implicit val jmc = new MinioClient(s"http://$host:$port", "12345678", "12345678")

  def stopSystem(): Future[Terminated] = mioSystem.terminate()
} 

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

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import edu.goldlok.minio_scala.auth.MioKeys

import scala.concurrent.ExecutionContextExecutor
import scala.concurrent.duration.FiniteDuration
import scala.concurrent.duration._


object TestElem {
  val EXPECTED_EXCEPTION_DID_NOT_FIRE = "Expected exception did not fire"
  val BUCKET = "bucket"
  val OBJECT = "object"
  val CONTENT_LENGTH = "Content-Length"
  val APPLICATION_OCTET_STREAM = "application/octet-stream"
  val APPLICATION_JAVASCRIPT = "application/javascript"
  val CONTENT_TYPE = "Content-Type"
  val MON_04_MAY_2015_07_58_51_GMT = "Mon, 04 May 2015 07:58:51 GMT"
  val LAST_MODIFIED = "Last-Modified"
  val HELLO_WORLD = "hello world"
  val HELLO = "hello"
  val BYTES = "bytes"
  val ENC_KEY = "x-amz-meta-x-amz-key"
  val ENC_IV = "x-amz-meta-x-amz-iv"
  val MAT_DESC = "x-amz-meta-x-amz-matdesc"
  val ACCEPT_RANGES = "Accept-Ranges"
  val CONTENT_RANGE = "Content-Range"
  val MON_29_JUN_2015_22_01_10_GMT = "Mon, 29 Jun 2015 22:01:10 GMT"
  val BUCKET_KEY = "/bucket/key"
  val MD5_HASH_STRING = "\"5eb63bbbe01eeed093cb22bb8f5acdc3\""

  implicit val mioSystem = ActorSystem("mio_system")
  implicit val mater =  ActorMaterializer()
  implicit val dispatcher: ExecutionContextExecutor = mioSystem.dispatcher
  implicit val keys = MioKeys("", "")
  val timeout: FiniteDuration = 180.seconds
} 

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

import akka.actor.ActorSystem
import org.helianto.hackabase.domain.Project
import play.api.libs.json._
import play.api.mvc._
import play.api.routing.sird._

import scala.concurrent.ExecutionContextExecutor

object HomeService {

  implicit val system: ActorSystem = ActorSystem("hackabase")
  implicit def executor: ExecutionContextExecutor = system.dispatcher

  val routes: PartialFunction[RequestHeader, Action[JsValue]] = {
    case POST(p"/sendConfirmation") => Action(BodyParsers.parse.json) { request =>
      request.body.validate[Project] match {
        case s: JsSuccess[Project] =>
          // TODO
          Results.Ok(s"Successfuly sent!")
        case e: JsError => Results.InternalServerError
      }
    }
  }


} 

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

import com.bisphone.cassandra
import com.bisphone.util.{AsyncResult, Convertor, SimpleError, ValueExtractor}
import com.bisphone.std._
import scala.concurrent.ExecutionContextExecutor


object Util {

  private val regex = {
    val raw = """(.*):(\d*)"""
    raw.r
  }

  implicit val tupleOfStringAndInt = Convertor[String, (String, Int)]("String => (String, Int)") {
    case regex(host,port) => (host, port.toInt)
    case x => throw new RuntimeException(s"Invalid String for 'host:port': '${x}'")
  }

  implicit val cassandraConsistencyLevel =
    Convertor[String, cassandra.ConsistencyLevel]("String => cassandra.ConsistencyLevel") { value =>
      cassandra.ConsistencyLevel.get(value) match {
        case Some(rsl) => rsl
        case None =>
          val msg = s"Invalid value for cassandra-consistencylevel: ${value}. Valid values: ${cassandra.ConsistencyLevel.values.mkString(",")}"
          throw new RuntimeException(msg)
      }
    }

  def cassandraConfig[T <: ValueExtractor](extractor: T)(
    implicit ex: ExecutionContextExecutor
  ): AsyncResult[SimpleError, cassandra.Config] = {
    for {
      seeds <- extractor.nelist[(String, Int)]("seeds")
      keyspace <- extractor.required[String]("keyspace")
      readCL <- extractor.required[cassandra.ConsistencyLevel]("read-consistency-level")
      writeCL <- extractor.required[cassandra.ConsistencyLevel]("write-consistency-level")
    } yield Config(seeds, keyspace, readCL, writeCL)
  }

  def cassandraConnection[T <: ValueExtractor](extractor: T)(
    implicit ex: ExecutionContextExecutor
  ): AsyncResult[SimpleError, cassandra.Connection] = {

    for {
      config <- cassandraConfig(extractor)

      conn = new cassandra.Connection(config)

    } yield conn
  }
} 

//设置package包名称以及导入依赖的类
package com.github.michalbogacz.http.streaming

import akka.NotUsed
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.{ContentTypes, HttpEntity}
import akka.http.scaladsl.server.Directives
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Source
import akka.util.ByteString
import com.mongodb.reactivestreams.client.{MongoClients, MongoCollection}
import org.bson.Document

import scala.concurrent.ExecutionContextExecutor
import scala.io.StdIn

object AppReturningChunking extends Directives {
  
  implicit val system = ActorSystem()
  implicit val dispatcher: ExecutionContextExecutor = system.dispatcher
  implicit val mat = ActorMaterializer()

  def main(args: Array[String]): Unit = {
    val mongoClient = MongoClients.create()
    val coll = mongoClient.getDatabase("test").getCollection("resources")

    val route =
      path("resources") {
        get {
          complete(HttpEntity(ContentTypes.`application/json`, getData(coll)))
        }
      }

    val bindingFuture = Http().bindAndHandle(route, "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.terminate()) // and shutdown when done

  }

  def getData(coll: MongoCollection[Document]): Source[ByteString, NotUsed] =
    Source.fromPublisher(coll.find())
      .map(_.toJson)
      .map(ByteString(_))
      .intersperse(ByteString("["), ByteString(","), ByteString("]"))

} 

//设置package包名称以及导入依赖的类
package com.github.michalbogacz.http.streaming

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.{ContentTypes, HttpEntity}
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Source
import com.mongodb.reactivestreams.client.{MongoClients, MongoCollection}
import org.bson.Document

import scala.concurrent.{ExecutionContextExecutor, Future}
import scala.io.StdIn

object ApiReturningList extends ServiceDirectives {

  implicit val system = ActorSystem()
  implicit val dispatcher: ExecutionContextExecutor = system.dispatcher
  implicit val mat = ActorMaterializer()

  def main(args: Array[String]): Unit = {
    val mongoClient = MongoClients.create()
    val coll = mongoClient.getDatabase("test").getCollection("resources")

    val route =
      path("resources") {
        pageParams { pageParams =>
          get {
            complete(getData(coll, pageParams).map(HttpEntity(ContentTypes.`application/json`, _)))
          }
        }
      }

    val bindingFuture = Http().bindAndHandle(route, "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.terminate()) // and shutdown when done

  }

  def getData(coll: MongoCollection[Document], pageParams: PageParams): Future[String] =
    Source.fromPublisher(coll.find().skip(pageParams.skip).limit(pageParams.limit))
      .map(_.toJson)
      .intersperse("[", ",", "]")
      .runFold("")((acc, e) ? acc + e)


} 

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

import java.util.concurrent.Executor

import reactor.core.Disposable
import reactor.core.scheduler.Scheduler.Worker
import reactor.core.scheduler.{Scheduler, Schedulers}

import scala.concurrent.{ExecutionContext, ExecutionContextExecutor, ExecutionContextExecutorService}

class ExecutionContextScheduler private(val scheduler: Scheduler) extends Scheduler {
  override def schedule(task: Runnable): Disposable = {
    val cancellation = scheduler.schedule(task)
    new Disposable {
      override def dispose(): Unit = cancellation.dispose()
    }
  }

  override def createWorker(): Worker = scheduler.createWorker()
}


object ExecutionContextScheduler {
  def apply(executionContext: ExecutionContext): ExecutionContextScheduler = {
    executionContext match {
      case eces: ExecutionContextExecutorService => new ExecutionContextScheduler(Schedulers.fromExecutorService(eces))
      case ece: ExecutionContextExecutor => new ExecutionContextScheduler(Schedulers.fromExecutor(ece))
      case _ => new ExecutionContextScheduler(Schedulers.fromExecutor(new Executor {
        override def execute(command: Runnable): Unit = executionContext.execute(command)
      }))
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.ubirch.avatar.core.actor

import akka.actor.{Actor, ActorLogging, Props}
import akka.http.scaladsl.HttpExt
import akka.stream.Materializer
import com.ubirch.avatar.core.device.DeviceDataRawManager
import com.ubirch.avatar.core.msgpack.MsgPacker
import com.ubirch.util.json.MyJsonProtocol
import com.ubirch.util.model.JsonErrorResponse
import com.ubirch.util.mongo.connection.MongoUtil
import org.apache.commons.codec.binary.Hex

import scala.concurrent.ExecutionContextExecutor


class MessageMsgPackProcessorActor(implicit mongo: MongoUtil, httpClient: HttpExt, materializer: Materializer)
  extends Actor
    with MyJsonProtocol


    with ActorLogging {

  implicit val executionContext: ExecutionContextExecutor = context.dispatcher

  private val validatorActor = context.system.actorOf(Props(new MessageValidatorActor()))
  //private val validatorActor = context.system.actorSelection(ActorNames.MSG_VALIDATOR)

  override def receive: Receive = {
    case binData: Array[Byte] =>
      val s = sender()

      val hexVal = Hex.encodeHexString(binData)
      log.info(s"got some MsgPack data: $hexVal")
      try {
        val (u, t) = MsgPacker.unpack(binData)
        DeviceDataRawManager.create(did = u, vals = t, mpraw = binData) match {
          case Some(drd) =>
            validatorActor forward drd
          case None =>
            log.error("could not parse input msgpack data")
            s ! JsonErrorResponse(errorType = "Validation Error", errorMessage = "Invalid Data")
        }
      }
      catch {
        case e: Exception =>
          log.error("received invalid data", e)
          sender ! JsonErrorResponse(errorType = "Invalid Data Error", errorMessage = "Incalid Dataformat")
      }
    case _ =>
      log.error("received unknown message")
      sender ! JsonErrorResponse(errorType = "Validation Error", errorMessage = "Invalid Input Data")
  }

} 

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

import akka.actor.{ActorLogging, ActorRef, Props}
import akka.camel.{CamelMessage, Consumer}
import akka.routing.RoundRobinPool
import com.ubirch.avatar.config.Config
import com.ubirch.avatar.core.device.DeviceManager
import com.ubirch.avatar.model.rest.device.DeviceDataRaw
import com.ubirch.avatar.util.actor.ActorNames
import com.ubirch.util.json.{Json4sUtil, MyJsonProtocol}

import scala.concurrent.ExecutionContextExecutor


class TransformerConsumerActor extends Consumer with ActorLogging with MyJsonProtocol {

  val accessKey: String = Config.awsAccessKey
  val secretKey: String = Config.awsSecretAccessKey

  override def endpointUri = s"aws-sqs://${Config.awsSqsQueueTransformer}?accessKey=$accessKey&secretKey=$secretKey&concurrentConsumers=2&maxMessagesPerPoll=10"

  override def autoAck: Boolean = true

  implicit val executionContext: ExecutionContextExecutor = context.dispatcher

  val transformerActor: ActorRef = context.actorOf(new RoundRobinPool(Config.akkaNumberOfWorkers).props(Props[TransformerPreprocessorActor]), ActorNames.TRANSFORMER_PRE)

  //TODO fix error handling, in case of error the message should be resend later?
  override def receive: Receive = {
    case msg: CamelMessage =>
      log.debug(s"received ${msg.bodyAs[String]}")
      msg.body match {
        case drdStr: String =>
          Json4sUtil.string2JValue(drdStr) match {
            case Some(drdJson) =>
              drdJson.extractOpt[DeviceDataRaw] match {
                case Some(drd) =>
                  DeviceManager.infoByHashedHwId(drd.a).map {
                    case Some(device) =>
                      transformerActor ! (device, drd)
                    case None =>
                      log.error(s"no device found for hashedHwdeviceId: ${drd.a}")
                  }
                case None =>
                  log.error(s"invalid json message from device: $drdStr")
              }
            case None =>
              log.error(s"invalid message from device: $drdStr")
          }

        case _ =>
          log.error(s"received invalid message body: ${msg.body}")
      }

    case _ =>
      log.error("received unknown message")
  }
} 

//设置package包名称以及导入依赖的类
package com.ubirch.avatar.backend.route

import com.typesafe.scalalogging.slf4j.StrictLogging

import com.ubirch.avatar.config.Config
import com.ubirch.avatar.core.device.DeviceManager
import com.ubirch.avatar.model.rest.device.DeviceInfo
import com.ubirch.avatar.util.server.RouteConstants._
import com.ubirch.util.http.response.ResponseUtil
import com.ubirch.util.oidc.directive.OidcDirective
import com.ubirch.util.rest.akka.directives.CORSDirective

import akka.actor.ActorSystem
import akka.http.scaladsl.server.Route
import akka.util.Timeout
import de.heikoseeberger.akkahttpjson4s.Json4sSupport._

import scala.concurrent.ExecutionContextExecutor
import scala.concurrent.duration._
import scala.language.postfixOps
import scala.util.{Failure, Success}


trait DeviceStubIdRoute extends ResponseUtil
    with CORSDirective
    with StrictLogging {

  implicit val system = ActorSystem()
  implicit val executionContext: ExecutionContextExecutor = system.dispatcher
  implicit val timeout = Timeout(Config.actorTimeout seconds)

  private val oidcDirective = new OidcDirective()

  val route: Route =

    path(stub / JavaUUID) { deviceId =>
      respondWithCORS {
        oidcDirective.oidcToken2UserContext { userContext =>
          get {
            onComplete(DeviceManager.stub(deviceId)) {
              case Success(resp) =>
                resp match {
                  case Some(stub: DeviceInfo) =>
                    complete(stub)
                  case _ =>
                    complete(requestErrorResponse(errorType = "QueryError", errorMessage = s"deviceId not found: deviceId=$deviceId"))
                }
              case Failure(t) =>
                logger.error("device creation failed", t)
                complete(serverErrorResponse(errorType = "DeviceStubError", errorMessage = t.getMessage))
            }

          }
        }
      }
    }
} 

//设置package包名称以及导入依赖的类
package com.ubirch.avatar.backend.route

import java.util.UUID

import akka.actor.ActorSystem
import akka.http.scaladsl.server.Route
import com.ubirch.avatar.core.avatar.AvatarStateManagerREST
import com.ubirch.avatar.core.device.DeviceManager
import com.ubirch.avatar.model.rest.device.AvatarState
import com.ubirch.avatar.util.server.RouteConstants._
import com.ubirch.util.http.response.ResponseUtil
import com.ubirch.util.mongo.connection.MongoUtil
import com.ubirch.util.oidc.directive.OidcDirective
import com.ubirch.util.rest.akka.directives.CORSDirective
import de.heikoseeberger.akkahttpjson4s.Json4sSupport._

import scala.concurrent.{ExecutionContextExecutor, Future}


class DeviceStateRoute(implicit mongo: MongoUtil)
  extends ResponseUtil
    with CORSDirective {

  implicit val system = ActorSystem()
  implicit val executionContext: ExecutionContextExecutor = system.dispatcher

  private val oidcDirective = new OidcDirective()

  val route: Route = {

    path(JavaUUID / state) { deviceId =>
      respondWithCORS {
        oidcDirective.oidcToken2UserContext { userContext =>
          get {
            onSuccess(queryState(deviceId)) {
              case None =>
                complete(requestErrorResponse(errorType = "QueryError", errorMessage = s"deviceId not found: deviceId=$deviceId"))
              case Some(avatarState: AvatarState) =>
                complete(avatarState)
            }
          }
        }
      }
    }

  }

  private def queryState(deviceId: UUID)(implicit mongo: MongoUtil): Future[Option[AvatarState]] = {
    DeviceManager.info(deviceId).flatMap {
      case Some(dvc) =>
        AvatarStateManagerREST.byDeviceId(dvc.deviceId)
      case None =>
        Future(None)
    }
  }

  //  private def storeState(deviceId: UUID, state: DeviceState): Future[Option[DeviceState]] = Future(None) // TODO implementation

} 

//设置package包名称以及导入依赖的类
package com.ubirch.avatar.backend.route

import akka.actor.{ActorSystem, Props}
import akka.http.scaladsl.HttpExt
import akka.http.scaladsl.server.{Directives, Route}
import akka.routing.RoundRobinPool
import akka.stream.Materializer
import akka.util.Timeout
import com.typesafe.scalalogging.slf4j.StrictLogging
import com.ubirch.avatar.config.Config
import com.ubirch.avatar.core.actor.MessageValidatorActor
import com.ubirch.avatar.model.rest.device.DeviceDataRaw
import com.ubirch.avatar.util.actor.ActorNames
import com.ubirch.avatar.util.server.RouteConstants._
import com.ubirch.util.http.response.ResponseUtil
import com.ubirch.util.model.JsonResponse
import com.ubirch.util.mongo.connection.MongoUtil

import scala.concurrent.ExecutionContextExecutor
import scala.concurrent.duration._
import scala.language.postfixOps
import de.heikoseeberger.akkahttpjson4s.Json4sSupport._


class DeviceUpdateBulkRoute(implicit mongo: MongoUtil, httpClient: HttpExt, materializer: Materializer)
  extends ResponseUtil
    with Directives
    with StrictLogging {

  implicit val system = ActorSystem()
  implicit val executionContext: ExecutionContextExecutor = system.dispatcher
  implicit val timeout = Timeout(Config.actorTimeout seconds)

  private val validatorActor = system.actorOf(new RoundRobinPool(Config.akkaNumberOfWorkers).props(Props(new MessageValidatorActor())), ActorNames.MSG_VALIDATOR)

  val route: Route = {

    path(update / bulk) {

      pathEnd {

        post {

          post {
            entity(as[DeviceDataRaw]) { sdm =>

              validatorActor ! sdm

              complete(JsonResponse(message = "processing started"))
            }
          }
        }
      }
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.flipkart.connekt.busybees.streams.flows.formaters

import com.fasterxml.jackson.databind.node.ObjectNode
import com.flipkart.connekt.busybees.streams.errors.ConnektPNStageException
import com.flipkart.connekt.busybees.streams.flows.NIOFlow
import com.flipkart.connekt.commons.entities.MobilePlatform
import com.flipkart.connekt.commons.factories.{ConnektLogger, LogFile, ServiceFactory}
import com.flipkart.connekt.commons.helpers.CallbackRecorder._
import com.flipkart.connekt.commons.helpers.ConnektRequestHelper._
import com.flipkart.connekt.commons.iomodels.MessageStatus.InternalStatus
import com.flipkart.connekt.commons.iomodels._
import com.flipkart.connekt.commons.services.DeviceDetailsService
import com.flipkart.connekt.commons.utils.StringUtils._

import scala.concurrent.ExecutionContextExecutor
import scala.concurrent.duration._

class AndroidChannelFormatter(parallelism: Int)(implicit ec: ExecutionContextExecutor) extends NIOFlow[ConnektRequest, GCMPayloadEnvelope](parallelism)(ec) {

  lazy val stencilService = ServiceFactory.getStencilService

  override def map: ConnektRequest => List[GCMPayloadEnvelope] = message => {

    try {
      ConnektLogger(LogFile.PROCESSORS).info(s"AndroidChannelFormatter received message: ${message.id}")
      ConnektLogger(LogFile.PROCESSORS).trace(s"AndroidChannelFormatter received message: ${message.toString}")

      val pnInfo = message.channelInfo.asInstanceOf[PNRequestInfo]

      val devicesInfo = DeviceDetailsService.get(pnInfo.appName, pnInfo.deviceIds).get.toSeq
      val validDeviceIds = devicesInfo.map(_.deviceId)
      val invalidDeviceIds = pnInfo.deviceIds.diff(validDeviceIds.toSet)

      invalidDeviceIds.map(PNCallbackEvent(message.id, message.clientId, _, InternalStatus.MissingDeviceInfo, MobilePlatform.ANDROID, pnInfo.appName, message.contextId.orEmpty)).persist
      ServiceFactory.getReportingService.recordPushStatsDelta(message.clientId, message.contextId, message.meta.get("stencilId").map(_.toString), Option(message.platform), message.appName, InternalStatus.MissingDeviceInfo, invalidDeviceIds.size)

      val tokens = devicesInfo.map(_.token)
      val androidStencil = stencilService.getStencilsByName(s"ckt-${pnInfo.appName.toLowerCase}-android").head
      val appDataWithId = stencilService.materialize(androidStencil, message.channelData.asInstanceOf[PNRequestData].data).asInstanceOf[String].getObj[ObjectNode]
        .put("messageId", message.id)
        .put("contextId", message.contextId.orEmpty)

      val ttl = message.expiryTs.map(expiry => (expiry - System.currentTimeMillis) / 1000).getOrElse(6.hour.toSeconds)

      if (tokens.nonEmpty && ttl > 0) {
        val payload = GCMPNPayload(registration_ids = tokens, delay_while_idle = Option(pnInfo.delayWhileIdle), appDataWithId, time_to_live = Some(ttl), dry_run = Option(message.isTestRequest))
        List(GCMPayloadEnvelope(message.id, message.clientId, validDeviceIds, pnInfo.appName, message.contextId.orEmpty, payload, message.meta))
      } else if (tokens.nonEmpty) {
        ConnektLogger(LogFile.PROCESSORS).warn(s"AndroidChannelFormatter dropping ttl-expired message: ${message.id}")
        devicesInfo.map(d => PNCallbackEvent(message.id, message.clientId, d.deviceId, InternalStatus.TTLExpired, MobilePlatform.ANDROID, d.appName, message.contextId.orEmpty)).persist
        ServiceFactory.getReportingService.recordPushStatsDelta(message.clientId, message.contextId, message.meta.get("stencilId").map(_.toString), Option(message.platform), message.appName, InternalStatus.TTLExpired, devicesInfo.size)
        List.empty[GCMPayloadEnvelope]
      } else
        List.empty[GCMPayloadEnvelope]
    } catch {
      case e: Exception =>
        ConnektLogger(LogFile.PROCESSORS).error(s"AndroidChannelFormatter error for ${message.id}", e)
        throw new ConnektPNStageException(message.id, message.clientId, message.destinations, InternalStatus.StageError, message.appName, message.platform, message.contextId.orEmpty, message.meta, "AndroidChannelFormatter::".concat(e.getMessage), e)
    }
  }
}