//设置package包名称以及导入依赖的类
package akka.actor.dungeon

import ReceiveTimeout.emptyReceiveTimeoutData
import akka.actor.ActorCell
import akka.actor.ActorCell.emptyCancellable
import akka.actor.Cancellable
import scala.concurrent.duration.Duration
import scala.concurrent.duration.FiniteDuration

private[akka] object ReceiveTimeout {
  final val emptyReceiveTimeoutData: (Duration, Cancellable) = (Duration.Undefined, ActorCell.emptyCancellable)
}

private[akka] trait ReceiveTimeout { this: ActorCell ?

  import ReceiveTimeout._
  import ActorCell._

  private var receiveTimeoutData: (Duration, Cancellable) = emptyReceiveTimeoutData

  final def receiveTimeout: Duration = receiveTimeoutData._1

  final def setReceiveTimeout(timeout: Duration): Unit = receiveTimeoutData = receiveTimeoutData.copy(_1 = timeout)

  final def checkReceiveTimeout() {
    val recvtimeout = receiveTimeoutData
    //Only reschedule if desired and there are currently no more messages to be processed
    if (!mailbox.hasMessages) recvtimeout._1 match {
      case f: FiniteDuration ?
        recvtimeout._2.cancel() //Cancel any ongoing future
        val task = system.scheduler.scheduleOnce(f, self, akka.actor.ReceiveTimeout)(this.dispatcher)
        receiveTimeoutData = (f, task)
      case _ ? cancelReceiveTimeout()
    }
    else cancelReceiveTimeout()

  }

  final def cancelReceiveTimeout(): Unit =
    if (receiveTimeoutData._2 ne emptyCancellable) {
      receiveTimeoutData._2.cancel()
      receiveTimeoutData = (receiveTimeoutData._1, emptyCancellable)
    }

} 

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

import java.time.{Clock, Instant}
import javax.inject._

import akka.actor.{ActorRef, ActorSystem, Cancellable}
import logic.actors.fakeactivity.FakeActivityActor.FakeActivity
import play.api.inject.ApplicationLifecycle
import play.api.{Logger, _}

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


@Singleton
class FakeActivity @Inject()(clock: Clock, appLifecycle: ApplicationLifecycle,
                             system: ActorSystem, @Named("fakeActivityActor") fakeActivityActor: ActorRef,
                             env: Environment) {

  // This code is called when the application starts.
  private val start: Instant = clock.instant
  private var cancellableFakeActivity: Cancellable = null


  Logger.info(s"FakeActivity: Starting application at ${start}.")

  Logger.info("Mode: " + env.mode)
  if (env.mode == Mode.Prod || env.mode == Mode.Dev) {
    cancellableFakeActivity = system.scheduler.schedule(3 second, 10 minutes, fakeActivityActor, FakeActivity)
  }


  // When the application starts, register a stop hook with the
  // ApplicationLifecyle object. The code inside the stop hook wil
  // be run when the application stops.
  appLifecycle.addStopHook { () =>
    val stop: Instant = clock.instant
    val runningTime: Long = stop.getEpochSecond - start.getEpochSecond
    Logger.info(s"FakeActivity: Stopping application at ${clock.instant} after ${runningTime}s.")
    if (cancellableFakeActivity != null) {
      cancellableFakeActivity.cancel()
    }

    Future.successful(())
  }
} 

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

import java.time.{Clock, Instant}
import javax.inject._

import akka.actor.{ActorRef, ActorSystem, Cancellable}
import play.api.Logger
import play.api._
import play.api.inject.ApplicationLifecycle

import scala.concurrent.Future
import logic.actors.schedule.CheckScheduleDateActor._

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


@Singleton
class ScheduleParser @Inject()(clock: Clock, appLifecycle: ApplicationLifecycle,system: ActorSystem , env: Environment,
                               @Named("checkSchedule") checkScheduleDateActor: ActorRef) {

  // This code is called when the application starts.
  private val start: Instant = clock.instant
  private var cancellable1:Cancellable = null



  Logger.info(s"ScheduleParser: Starting application at ${start}.")

  if(env.mode == Mode.Prod || env.mode == Mode.Dev) {
    cancellable1 = system.scheduler.schedule(2 second, 1 day ,checkScheduleDateActor , CheckScheduleDate)
  }

  // When the application starts, register a stop hook with the
  // ApplicationLifecyle object. The code inside the stop hook wil
  // be run when the application stops.
  appLifecycle.addStopHook { () =>
    val stop: Instant = clock.instant
    val runningTime: Long = stop.getEpochSecond - start.getEpochSecond
    Logger.info(s"ScheduleParser: Stopping application at ${clock.instant} after ${runningTime}s.")
    if(cancellable1 != null){
      cancellable1.cancel()
    }
    Future.successful(())
  }

} 

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

import java.time.{Clock, Instant}
import javax.inject._

import akka.actor.{ActorRef, ActorSystem, Cancellable}
import logic.actors.rss.DeleteNewsActor._
import logic.actors.rss.NewsReaderActor._
import play.api.{Logger, _}
import play.api.inject.ApplicationLifecycle

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


@Singleton
class NewsFeedReader @Inject()(clock: Clock, appLifecycle: ApplicationLifecycle,
                               system: ActorSystem, @Named("NewsReader") newsReaderActor: ActorRef,
                               @Named("deleteNews") deleteNewsActor: ActorRef, env: Environment) {

  // This code is called when the application starts.
  private val start: Instant = clock.instant
  private var cancellableFeedParser: Cancellable = null
  private var cancellableNewsDeleter: Cancellable = null

  Logger.info(s"NewsFeedReader: Starting application at ${start}.")

  Logger.info("Mode: " + env.mode)
  if (env.mode == Mode.Prod || env.mode == Mode.Dev) {
    cancellableFeedParser = system.scheduler.schedule(3 second, 1 minute, newsReaderActor, ReadNews)
    cancellableNewsDeleter = system.scheduler.schedule(10 seconds, 6 hours, deleteNewsActor, DeleteNews)
  }


  // When the application starts, register a stop hook with the
  // ApplicationLifecyle object. The code inside the stop hook wil
  // be run when the application stops.
  appLifecycle.addStopHook { () =>
    val stop: Instant = clock.instant
    val runningTime: Long = stop.getEpochSecond - start.getEpochSecond
    Logger.info(s"NewsFeedReader: Stopping application at ${clock.instant} after ${runningTime}s.")
    if (cancellableFeedParser != null) {
      cancellableFeedParser.cancel()
    }
    if (cancellableNewsDeleter != null) {
      cancellableNewsDeleter.cancel()
    }
    Future.successful(())
  }
} 

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

import java.time.{Clock, Instant}
import javax.inject._

import akka.actor.{ActorRef, ActorSystem, Cancellable}
import logic.actors.database.DatabaseActor.{StartUp, SyncDatabase}
import logic.actors.fakeactivity.FakeActivityActor.FakeActivity
import play.api.inject.ApplicationLifecycle
import play.api.{Logger, _}

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


@Singleton
class DatabaseService @Inject()(clock: Clock, appLifecycle: ApplicationLifecycle,
                                system: ActorSystem, @Named("databaseActor") databaseActor: ActorRef,
                                env: Environment) {

  // This code is called when the application starts.
  private val start: Instant = clock.instant
  private var cancellableSyncActivity: Cancellable = null


  Logger.info(s"Database: Starting application at ${start}.")

  Logger.info("Mode: " + env.mode)
  if (env.mode == Mode.Prod || env.mode == Mode.Dev) {
    system.scheduler.scheduleOnce(1 second, databaseActor, StartUp)
    cancellableSyncActivity = system.scheduler.schedule(30 second, 15 minutes, databaseActor, SyncDatabase)
  }


  // When the application starts, register a stop hook with the
  // ApplicationLifecyle object. The code inside the stop hook wil
  // be run when the application stops.
  appLifecycle.addStopHook { () =>
    val stop: Instant = clock.instant
    val runningTime: Long = stop.getEpochSecond - start.getEpochSecond
    Logger.info(s"Database: Stopping application at ${clock.instant} after ${runningTime}s.")
    if (cancellableSyncActivity != null) {
      cancellableSyncActivity.cancel()
    }

    Future.successful(())
  }
} 

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

import java.util.concurrent.TimeUnit

import akka.actor.{ActorRef, Cancellable}
import com.init6.Init6Actor
import com.init6.users.KillConnection

import scala.concurrent.duration.Duration


trait Init6KeepAliveActor extends Init6Actor {

  private var pingTask: Cancellable = _
  protected var keptAlive = 0

  override def postStop(): Unit = {
    Option(pingTask).foreach(_.cancel())

    super.postStop()
  }

  def keepAlive(actor: ActorRef, f: () => Unit): Unit = {
    keepAlive(actor, f, 25, TimeUnit.SECONDS)
  }

  def keepAlive(actor: ActorRef, f: () => Unit, delay: Long, unit: TimeUnit): Unit = {
    val pingDuration = Duration(25, TimeUnit.SECONDS)
    import context.dispatcher

    pingTask = system.scheduler.schedule(
      pingDuration,
      pingDuration
    )({
        if (keptAlive < 4) {
          keptAlive += 1
          f()
        } else {
          actor ! KillConnection
        }
    })
  }
} 

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

import java.util.concurrent.TimeUnit

import akka.actor.{Cancellable, Props}
import com.init6.Constants._
import com.init6.coders.commands.{BroadcastCommand, Command}
import com.init6.{Config, Init6Actor, Init6Component}

import scala.concurrent.duration.{Duration, FiniteDuration}


object ServerAnnouncementActor extends Init6Component {
  def apply(timeToDrop: Long) =
    system.actorOf(Props(classOf[ServerAnnouncementActor], timeToDrop), INIT6_SERVER_ANNOUNCEMENT_PATH)
}

case class RepeatingAnnoucement(message: String, duration: FiniteDuration) extends Command

class ServerAnnouncementActor(timeToDrop: Long) extends Init6Actor {

  var announcement: Option[Cancellable] = None

  override def preStart() = {
    super.preStart()

    if (!Config().Server.Chat.enabled) {
      import context.dispatcher
      system.scheduler.scheduleOnce(Duration(
        timeToDrop - 15, TimeUnit.SECONDS
      ))({
        usersActor ! BroadcastCommand(WILL_DROP_IN(Config().Server.host, 15))
      })
    }
  }

  override def receive = {
    case RepeatingAnnoucement(message, duration) =>
      import context.dispatcher
      if (announcement.isDefined) {
        announcement.get.cancel()
      }

      announcement = Some(system.scheduler.schedule(
        Duration(0, TimeUnit.MILLISECONDS),
        duration
      )({
        usersActor ! BroadcastCommand(message)
      }))
  }
} 

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

import akka.actor._
import akka.event.LoggingReceive
import play.api.Logger
import play.api.libs.iteratee.Concurrent.Channel
import prickle.Pickle
import shared._
import shared.Utils.currentSeconds
import akka.actor.Cancellable
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._



  val keepAlive : Cancellable = ActorSystem(s"keepAlive").scheduler.schedule(0.seconds, SEND_INTERVAL){
    backChannel push Pickle.intoString(CKeepAlive() : NetworkMessage)
    if (currentSeconds - lastResponseTime >= CONNECTION_TIMEOUT_SECONDS) {
      keepAlive.cancel()
      mainServerActor ! CUserDisconnected()
      //this prevents that in the last run there could be an nullPointer
      if(context != null && self != null){
        Logger.error(s"My Client ${context.self.toString()} didn't send anything for ${currentSeconds - lastResponseTime}s...")
        context.stop(self)
      }
    }
  }

  def receive = LoggingReceive {
    //Messages to the client go out through the backChannel
    case n: ServerToClient =>
      Logger.error("ServerConnectionActor-toClient-" + n)
      backChannel push Pickle.intoString(n: NetworkMessage)

    //Messages from the client go the the mainServerActor
    case m: ClientToServer =>
      Logger.error("ServerConnectionActor-fromClient-" + m)
      m match {
        case CKeepAlive() => lastResponseTime = currentSeconds
        case _            => mainServerActor ! m
      }

    case x => Logger.error("Unknown packet in ConnectionHandlerActor - " + x)
  }

}

object ServerConnectionActor {
  def props(backChannel: Channel[String], mainServerActor: ActorRef): Props = Props(new ServerConnectionActor(backChannel, mainServerActor))
} 

//设置package包名称以及导入依赖的类
package akka.actor.dungeon

import ReceiveTimeout.emptyReceiveTimeoutData
import akka.actor.ActorCell
import akka.actor.ActorCell.emptyCancellable
import akka.actor.Cancellable
import scala.concurrent.duration.Duration
import scala.concurrent.duration.FiniteDuration

private[akka] object ReceiveTimeout {
  final val emptyReceiveTimeoutData: (Duration, Cancellable) = (Duration.Undefined, ActorCell.emptyCancellable)
}

private[akka] trait ReceiveTimeout { this: ActorCell ?

  import ReceiveTimeout._
  import ActorCell._

  private var receiveTimeoutData: (Duration, Cancellable) = emptyReceiveTimeoutData

  final def receiveTimeout: Duration = receiveTimeoutData._1

  final def setReceiveTimeout(timeout: Duration): Unit = receiveTimeoutData = receiveTimeoutData.copy(_1 = timeout)

  final def checkReceiveTimeout() {
    val recvtimeout = receiveTimeoutData
    //Only reschedule if desired and there are currently no more messages to be processed
    if (!mailbox.hasMessages) recvtimeout._1 match {
      case f: FiniteDuration ?
        recvtimeout._2.cancel() //Cancel any ongoing future
      val task = system.scheduler.scheduleOnce(f, self, akka.actor.ReceiveTimeout)(this.dispatcher)
        receiveTimeoutData = (f, task)
      case _ ? cancelReceiveTimeout()
    }
    else cancelReceiveTimeout()

  }

  final def cancelReceiveTimeout(): Unit =
    if (receiveTimeoutData._2 ne emptyCancellable) {
      receiveTimeoutData._2.cancel()
      receiveTimeoutData = (receiveTimeoutData._1, emptyCancellable)
    }

} 

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

import scala.concurrent.duration.FiniteDuration

import scala.scalajs.js.timers.{SetIntervalHandle, setInterval, clearInterval}
import akka.actor.Cancellable

class JSIntervalTask(interval: FiniteDuration, task: => Any) extends Cancellable {
  private[this] var underlying: Option[SetIntervalHandle] =
    Some(setInterval(interval)(task))

  def isCancelled: Boolean = underlying.isEmpty

  def cancel(): Boolean = {
    if (isCancelled) false
    else {
      clearInterval(underlying.get)
      underlying = None
      true
    }
  }
}

object JSIntervalTask {
  def apply(interval: FiniteDuration)(task: => Any): JSIntervalTask =
    new JSIntervalTask(interval, task)
} 

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

import scala.concurrent.duration.{Duration, FiniteDuration}

import akka.actor.Cancellable

class JSTimeoutThenIntervalTask(initialDelay: FiniteDuration,
    interval: FiniteDuration, task: => Any) extends Cancellable {

  private[this] var underlying: Cancellable = JSTimeoutTask(initialDelay) {
    underlying = JSIntervalTask(interval) {
      task
    }
    task
  }

  def isCancelled: Boolean = underlying.isCancelled

  def cancel(): Boolean = underlying.cancel()
}

object JSTimeoutThenIntervalTask {
  def apply(initialDelay: FiniteDuration, interval: FiniteDuration)(
      task: => Any): JSTimeoutThenIntervalTask =
    new JSTimeoutThenIntervalTask(initialDelay, interval, task)
} 

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

import scala.concurrent.duration.FiniteDuration

import scala.scalajs.js.timers.{SetTimeoutHandle, setTimeout, clearTimeout}
import akka.actor.Cancellable

class JSTimeoutTask(delay: FiniteDuration, task: => Any) extends Cancellable {
  private[this] var underlying: Option[SetTimeoutHandle] =
    Some(setTimeout(delay)(task))

  def isCancelled: Boolean = underlying.isEmpty

  def cancel(): Boolean = {
    if (isCancelled) false
    else {
      clearTimeout(underlying.get)
      underlying = None
      true
    }
  }
}

object JSTimeoutTask {
  def apply(duration: FiniteDuration)(task: => Any): JSTimeoutTask =
    new JSTimeoutTask(duration, task)
} 

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

import java.net.URI

import akka.actor.{ActorRef, ActorSystem, Cancellable}
import bridgeapp.crawler.execution._
import com.typesafe.scalalogging.LazyLogging

import scala.concurrent.duration._
import scala.util.control.NonFatal


class RobotsTxtLookUp(throttling: ActorRef) extends Runnable with LazyLogging {
  override def run(): Unit = {
    logger.warn(s" Lookup robots.txt")
    val uri = new URI("https://rutracker.org/robots.txt")

    val parser = new ResponseParser {
      override def ->(response: Response): Unit = {
        val body = new String(response.body)
        Robots.parse(body) match {
          case Left(error) =>
            logger.error(s"Error parse robots txt file, $error ")
          case Right(robots) =>
            robots.nonGroupFields.map(s => (s.key, s.value)).toMap[String, String].get("Crawl-delay").fold() { delay =>
              try {
                val rate = Rate(1, delay.toFloat.second)
                throttling ! SetRate(rate)
              } catch {
                case NonFatal(e) => logger.error("Error parse crawl delay directive in robots.txt. ", e)
              }
            }
        }
      }
    }
    val request = Request(uri, parser)

    //throttling !! request
  }
}

class RobotsTXTScheduleExecutor(robotsTxtLookUp: RobotsTxtLookUp)(implicit val actorSystem: ActorSystem) {

  private implicit val prep = actorSystem.dispatcher.prepare()

  private var instance: Cancellable = _

  def execute(duration: FiniteDuration) = {
    instance = actorSystem.scheduler.schedule(duration, duration, robotsTxtLookUp)
  }

  def cancel(): Unit = {
    instance.cancel()
  }
} 

//设置package包名称以及导入依赖的类
package com.outr.arango.managed

import akka.actor.{ActorSystem, Cancellable, Terminated}
import com.outr.arango.rest.LogEvent
import reactify.Observable

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

class RealTime(graph: Graph) {
  private lazy val system = ActorSystem("Graph_realTime")
  private var cancellable: Option[Cancellable] = None

  lazy val events: Observable[LogEvent] = graph.monitor

  def start(delay: FiniteDuration = 250.millis): Unit = synchronized {
    assert(cancellable.isEmpty, "Graph.realTime is already started.")
    cancellable = Some(system.scheduler.schedule(delay, delay) {
      update()
    })
  }

  def update(): Unit = graph.monitor.updateAndWait()

  def stop(): Unit = {
    cancellable.foreach(_.cancel())
    cancellable = None
  }

  def started: Boolean = cancellable.nonEmpty

  def shutdown(): Future[Terminated] = synchronized {
    stop()
    system.terminate()
  }
} 

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

import akka.actor.{Actor, ActorLogging, ActorRef, Cancellable}
import akka.io.Tcp.{PeerClosed, Write}
import akka.util.ByteString
import org.mystic.model._

class WriteHandler(whereToSend: ActorRef) extends Actor with ActorLogging {

  var cancelable: Cancellable = _

  override def receive: Receive = {

    case Last10Minutes =>
      log.info("trying to send 10 minutes data to the client")
      val storage = context.system.actorSelection("user/storage")
      storage ! AskFor10MData

    case Data(data: List[Option[CandleDeal]]) =>
      data.filter(_.isDefined).map(_.get).map(_.toJson).foreach(x => {
        whereToSend ! Write(ByteString(x))
      })

    case Cancel(cancel) =>
      log.info("setting the cancellable")
      cancelable = cancel

    case Last1Minute =>
      log.info("trying to send last minute data to the client")
      val storage = context.system.actorSelection("user/storage")
      storage ! AskFor1MData

    case PeerClosed =>
      log.info("peer closed")
      cancelable.cancel()
      context stop self

    case _ => log.error("something goes wrong in WriteHandler")
  }

} 

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

import akka.actor.{ Actor, ActorRef, Cancellable }
import workers.config.NewsConfig

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

class PollingScheduler(
    receiver: ActorRef,
    config: NewsConfig
) extends Actor {
  override def receive: Receive = PartialFunction.empty

  
  val polling: Cancellable = context.system.scheduler.schedule(
    0.seconds,
    config.pollingInterval,
    receiver,
    MessageReceiver.Receive
  )

  @scala.throws[Exception](classOf[Exception])
  override def preRestart(reason: Throwable, message: Option[Any]): Unit = {
    if (!polling.isCancelled) {
      polling.cancel()
    }
    super.preRestart(reason, message)
  }
} 

//设置package包名称以及导入依赖的类
package com.github.kelebra.akka.js.snake

import akka.actor.{Actor, ActorRef, Cancellable, Terminated}

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

case class Game(snake: ActorRef,
                pane: ActorRef,
                `random block generator`: () => Block,
                `moves per second`: Int) extends Actor {

  private val `millis per move`: FiniteDuration = (1000 milliseconds) / `moves per second`

  def receive: Receive = behavior()

  private def behavior(movement: Option[Cancellable] = None): Receive = {
    case start: Start =>
      context watch snake
      snake forward start
      self ! Fruitless
      context.become(
        behavior(Option(context.system.scheduler.schedule(`millis per move`, `millis per move`, snake, Move)))
      )
    case Fruitless =>
      val fruit = `random block generator`()
      pane ! Draw(fruit)
      snake ! Fruit(fruit)
    case _: Terminated =>
      movement.foreach(_.cancel())
      context unwatch snake
  }
} 

//设置package包名称以及导入依赖的类
package mesosphere.marathon.core.flow.impl

import akka.actor.{ Actor, Cancellable, Props }
import mesosphere.marathon.core.flow.LaunchTokenConfig
import mesosphere.marathon.core.instance.update.{ InstanceChange, InstanceUpdated }
import mesosphere.marathon.core.matcher.manager.OfferMatcherManager
import mesosphere.marathon.core.task.bus.TaskChangeObservables
import rx.lang.scala.{ Observable, Subscription }

import scala.concurrent.duration._

private[flow] object OfferMatcherLaunchTokensActor {
  def props(
    conf: LaunchTokenConfig,
    taskStatusObservables: TaskChangeObservables,
    offerMatcherManager: OfferMatcherManager): Props = {
    Props(new OfferMatcherLaunchTokensActor(conf, taskStatusObservables, offerMatcherManager))
  }
}


private class OfferMatcherLaunchTokensActor(
  conf: LaunchTokenConfig,
  taskStatusObservables: TaskChangeObservables, offerMatcherManager: OfferMatcherManager)
    extends Actor {
  var taskStatusUpdateSubscription: Subscription = _
  var periodicSetToken: Cancellable = _

  override def preStart(): Unit = {
    val all: Observable[InstanceChange] = taskStatusObservables.forAll
    taskStatusUpdateSubscription = all.subscribe(self ! _)

    import context.dispatcher
    periodicSetToken = context.system.scheduler.schedule(0.seconds, conf.launchTokenRefreshInterval().millis)(
      offerMatcherManager.setLaunchTokens(conf.launchTokens())
    )
  }

  override def postStop(): Unit = {
    taskStatusUpdateSubscription.unsubscribe()
    periodicSetToken.cancel()
  }

  override def receive: Receive = {
    case InstanceUpdated(instance, _, _) if instance.isRunning && instance.state.healthy.fold(true)(_ == true) =>
      offerMatcherManager.addLaunchTokens(1)
  }
} 

//设置package包名称以及导入依赖的类
import akka.actor.{Actor, Cancellable, ActorRef, Props}
import UnitsScraper.{Scan, SearchResults}
import scala.collection.mutable.{HashMap, HashSet}
import scala.concurrent.duration._

//TODO: schedule checkup/removal at endtime. Future: bid-bot
class ScraperController extends Actor { //TODO: implement persistence in case of crash
  import ScraperController._ //case classes defined in companion object
  val searchCancelActorMap = new HashMap[SearchTerm, (ActorRef, Cancellable)]
  val oldCategorizedResults = new HashMap[SearchTerm, Set[AuctionObject]]
  val oldObjects = new HashSet[AuctionObject]
  val serverActor = context.actorOf(Props[WebServer], name = "Server")
  val pushbullet = context.actorOf(Props[PushAPI], name = "Pushbullet")
  val system = akka.actor.ActorSystem("system")

  def receive = {
    case SearchResults(result) => {
      val searchTerm = sender.path.name
      val newObjectsInCategory = result -- oldCategorizedResults.getOrElse(SearchTerm(searchTerm), new HashSet[AuctionObject])
      serverActor ! NewCategoryResults(searchTerm, newObjectsInCategory)
      if (Preferences.isPushbulletActive && !Preferences.NeedtoConfigure) {
        for (obj <- newObjectsInCategory) {
          pushbullet ! pushLink(s"""New auction object in search "$searchTerm": ${obj.title}""", obj.link)
        }
      }
      oldCategorizedResults(SearchTerm(searchTerm)) = result
    }
    case NewSearchTerm(s) => {
      val sT = SearchTerm(s)
      //creates an actor in this context, i.e. a child actor
      val actor = context.actorOf(Props[UnitsScraper], name = s)
      //sets the ExecutionContext for the message sending task
      import system.dispatcher
      //sets a schedule to scan the search page every 5 minutes, starting in 200 milliseconds
      val cancel: Cancellable = system.scheduler.schedule(200 milliseconds, 5 minutes, actor, Scan)(sender = context.self, executor = system.dispatcher)
      searchCancelActorMap += (sT -> ((actor, cancel)))
    }
    case DeleteSearchTerm(s) => {
      val sT = SearchTerm(s)
      val actorCancelTuple = searchCancelActorMap(sT)
      actorCancelTuple._2.cancel
      context stop actorCancelTuple._1
      searchCancelActorMap -= sT
    }
    case WebServer.Shutdown => context.system.terminate
  }
}

object ScraperController {
  case class SearchTerm(s: String)
  case class NewSearchTerm(s: String)
  case class DeleteSearchTerm(s: String)
  case class NewCategoryResults(searchTerm: String, objects: Set[AuctionObject])
} 

//设置package包名称以及导入依赖的类
package com.stacktrace.yo.scrapeline.old

import akka.actor.{Actor, ActorLogging, Cancellable, Props}
import com.stacktrace.yo.scrapeline.old.HttpRequestSupervisor.SendNextRequests
import com.stacktrace.yo.scrapeline.old.ScrapeActor.{BeginScrape, ScrapeContent}
import net.ruippeixotog.scalascraper.model.Document

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

class HttpRequestSupervisor(val urlSet: Set[String]) extends Actor with ActorLogging {

  import context.dispatcher

  val maxRetries = 2
  var numVisited = 0
  var inProcess = 0
  val toProcess = scala.collection.mutable.Queue(urlSet.toList: _*)

  val tick: Cancellable = context.system.scheduler.schedule(0 millis, 1000 millis, self, SendNextRequests())

  override def receive: Receive = {
    case SendNextRequests() =>
      if (numVisited < urlSet.size) {
        val sendOut = Math.min(toProcess.size, 10 - inProcess)
        for (i <- 1 to sendOut) {
          val url = toProcess.dequeue
          val reader = context.actorOf(Props(new ScrapeActor()))
          reader ! BeginScrape(url)
          inProcess += 1
        }
      }
    case ScrapeContent(document: Document) => {
      inProcess -= 1
      numVisited += 1
      log.info("Recieved Response {} left in queue, {} in process", toProcess.size, inProcess)

    }
  }
}

object HttpRequestSupervisor {

  case class SendNextRequests()

}