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

import java.net.URL

import akka.actor.{ Actor, ActorSystem, Props, _ }
import au.csiro.data61.magda.external.InterfaceConfig
import com.typesafe.config.Config

import scala.language.postfixOps

class Supervisor(system: ActorSystem, config: Config, val externalInterfaces: Seq[InterfaceConfig]) extends Actor with ActorLogging {
  val indexer = context actorOf Props(new Indexer(self))
  val host2Actor: Map[URL, ActorRef] = externalInterfaces
    .groupBy(_.baseUrl)
    .mapValues(interfaceConfig => system.actorOf(Props(new RepoCrawler(self, indexer, interfaceConfig.head))))

  def receive: Receive = {
    case ScrapeAll =>
      log.info("Beginning scrape with interfaces {}", host2Actor.map { case (url, _) => url.toString })
      host2Actor.foreach { case (_, interface) => interface ! ScrapeRepo }
    case NeedsReIndexing =>
      log.info("Search index is empty, commanding crawlers to rebuild it ??????")
      self ! ScrapeAll
    case ScrapeRepoFailed(baseUrl, reason) =>
      log.error(reason, "Failed to start index for {} due to {}", baseUrl, reason.getMessage)
    case ScrapeRepoFinished(baseUrl) =>
      log.info("Finished scraping {}", baseUrl)
    case IndexFinished(dataSets, baseUrl) =>
      log.info("Finished indexing {} datasets from {}", dataSets.length, baseUrl)
    case IndexFailed(baseUrl, reason) =>
      log.error(reason, "Failed to index datasets for {} due to {}", baseUrl, reason.getMessage)
  }
} 

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

import javax.inject._

import akka.actor.{ActorSystem, _}
import akka.stream.Materializer
import play.api.libs.json._
import play.api.libs.streams.ActorFlow
import play.api.mvc._
import models._
import play.api.cache.{Cache, CacheApi}



@Singleton
class WebSocketController @Inject()(implicit system: ActorSystem, materializer: Materializer, cache: CacheApi) {

  object MyWebSocketActor {
    def props(out: ActorRef) = Props(new MyWebSocketActor(out))
  }

  class MyWebSocketActor(out: ActorRef) extends Actor {
    override def preStart() {
      println("open")
      println(self.path.parent.toSerializationFormat)
      val x = system.actorSelection(self.path.parent.toSerializationFormat)
      x ! "Message12"
    }

    override def postStop() {
      println("close")
    }

    override def receive = {
      case request: JsValue =>
        val response = handleMessage(request)
        out ! response
    }

    def handleMessage(event: Event): JsValue = {
      event match {
        case event: EventA => {
          val actorId = self.path.parent.toSerializationFormat
          cache.set("actor-id_" + event.data, actorId)
          Json.toJson(event)
        }
        case event: EventB => Json.toJson(event)
      }
    }
  }

  def socket = WebSocket.accept[JsValue, JsValue] { request =>
    ActorFlow.actorRef(out => MyWebSocketActor.props(out))

  }

} 

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

import javax.inject._

import akka.actor.{ActorSystem, _}
import dao.CatDAO
import models._
import play.api.cache.CacheApi
import play.api.data.Form
import play.api.data.Forms.{mapping, text}
import play.api.libs.concurrent.Execution.Implicits.defaultContext
import play.api.libs.json.Json
import play.api.mvc._



@Singleton
class HomeController @Inject()(implicit system: ActorSystem, catDao: CatDAO, cache: CacheApi) extends Controller {

  def index = Action.async {
    val actorId = cache.get[String]("actor-id_1").get
    println(actorId)
    val client = system.actorSelection(actorId)
    client ! Json.toJson(EventB("MessageB"))

    catDao.all().map {
      cats => Ok(views.html.index(cats))
    }
  }

  def insertCat = Action.async { implicit request =>
    val cat: Cat = catForm.bindFromRequest.get
    catDao.insert(cat).map(_ => Redirect(routes.HomeController.index))
  }

  def ws = Action { request => Ok(views.html.ws()) }

  val catForm = Form(
    mapping(
      "name" -> text(),
      "color" -> text()
    )(Cat.apply)(Cat.unapply)
  )
} 

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

import akka.actor.{Actor, ActorRef, ActorSystem, _}

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._

case class StartDetails(rtSite: RTSite,
                        rTCategory: RTCategory)

case class StoreDetails(details: RTItem,
                        rtSite: RTSite,
                        rTCategory: RTCategory)

case class EndDetails()

case class ScrapDetails(url: String,
                        rtSite: RTSite,
                        rTCategory: RTCategory)

class Supervisor(system: ActorSystem) extends Actor {

  var host2Actor = Map.empty[String, ActorRef]

  val scrapers = system.actorOf(Props(new Scraper(self)))

  override def receive: Receive = {
    case StartDetails(rtSite: RTSite, rTCategory: RTCategory) => startDetails(rtSite, rTCategory)
    case StoreDetails(details: RTItem, rtSite: RTSite, rtCategory: RTCategory) =>
      storeDetails(details, rtSite, rtCategory)
    case EndDetails => ???
  }

  private def storeDetails(details: RTItem,
                           rtSite: RTSite,
                           rtCategory: RTCategory) = {
    Store.writeDetails(details, rtSite, rtCategory)
  }

  private def startDetails(site: RTSite,
                           category: RTCategory) = {

    //    var id = 2270417
    var id = 2284843
    val last = 2271641
    //    val finish = 2280415
    system.scheduler.schedule(2 seconds, 2 seconds)({
      scrapers ! ScrapDetails(
        s"https://en.aruodas.lt/1-$id/",
        site,
        category
      )
      id = id + 1
    })

  }


} 

//设置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 services

import java.time.temporal.ChronoUnit
import java.time.{Clock, LocalDateTime}
import javax.inject.Inject

import akka.actor.{Actor, _}
import akka.pattern.pipe
import common.Email
import model.GmailWatchReply
import play.api.Logger
import services.GmailWatcherActor._

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

object GmailWatcherActor {
  final val actorName = "gmailWatcherActor"
  case class StartWatch(email: Email)
  case class WatchResponsePacket(email: Email, gmailWatchReply: GmailWatchReply, client: ActorRef)
  case class StartWatchDone(email: Email, timeBeforeRenew: Long)
}

class GmailWatcherActor @Inject()(clock: Clock, mailClient: MailClient) extends Actor with ActorLogging {
  implicit val executor = context.dispatcher

  var renew: Option[akka.actor.Cancellable] = None

  override def receive: Receive = {
    case StartWatch(email) =>
      Logger.info(s"StartWatch for $email")
      val s = sender
      pipe(mailClient.watch(email, List(LabelService.InboxLabelName)).map(WatchResponsePacket(email, _, s))) to self
    case WatchResponsePacket(email, gmailWatchReply, client) =>
      val now = LocalDateTime.now(clock)
      val timeBeforeRenew = now.until(gmailWatchReply.expirationDate, ChronoUnit.SECONDS)
      renew = Some(context.system.scheduler.scheduleOnce(timeBeforeRenew seconds)(self ! StartWatch(email)))
      client ! StartWatchDone(email, timeBeforeRenew)
      Logger.info(s"WatchResponse received for $email with expiration date ${gmailWatchReply.expirationDate} and historyId ${gmailWatchReply.historyId}")
    case Status.Failure(e) =>
      Logger.error(s"Watch failed", e)
      context.stop(self)
  }

  override def postStop(): Unit = {
    renew.fold(())(_.cancel)
  }
}