//设置package包名称以及导入依赖的类
package com.pacbio.secondary.smrtlink.actors

import java.util.UUID

import akka.actor.{Props, ActorRef, Actor}
import com.pacbio.common.actors.{PacBioActor, ActorRefFactoryProvider}
import com.pacbio.common.dependency.Singleton
import com.pacbio.secondary.smrtlink.models.{RegistryResourceUpdate, RegistryProxyRequest, RegistryResourceCreate}

// TODO(smcclellan): Add scaladoc

object RegistryServiceActor {
  case class GetResources(id: Option[String])
  case class GetResource(uuid: UUID)
  case class CreateResource(create: RegistryResourceCreate)
  case class UpdateResource(uuid: UUID, update: RegistryResourceUpdate)
  case class DeleteResource(uuid: UUID)
  case class ProxyRequest(uuid: UUID, req: RegistryProxyRequest)
}

class RegistryServiceActor(registryDao: RegistryDao) extends PacBioActor {
  import RegistryServiceActor._

  def receive: Receive = {
    case GetResources(id)             => respondWith(registryDao.getResources(id))
    case GetResource(uuid)            => respondWith(registryDao.getResource(uuid))
    case CreateResource(create)       => respondWith(registryDao.createResource(create))
    case UpdateResource(uuid, update) => respondWith(registryDao.updateResource(uuid, update))
    case DeleteResource(uuid)         => respondWith(registryDao.deleteResource(uuid))
    case ProxyRequest(uuid, req)      => respondWith(registryDao.proxyRequest(uuid, req))
  }
}

trait RegistryServiceActorRefProvider {
  this: RegistryDaoProvider with ActorRefFactoryProvider =>

  val registryServiceActorRef: Singleton[ActorRef] =
    Singleton(() => actorRefFactory().actorOf(Props(classOf[RegistryServiceActor], registryDao()), "RegistryServiceActor"))
}

trait RegistryServiceActorProvider {
  this: RegistryDaoProvider =>

  val registryServiceActor: Singleton[RegistryServiceActor] = Singleton(() => new RegistryServiceActor(registryDao()))
} 

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

import java.util.concurrent.{ CountDownLatch, TimeUnit }

import akka.testkit._
import akka.actor.{ Props, Actor }
import akka.testkit.AkkaSpec
import org.scalatest.BeforeAndAfterEach
import akka.dispatch.{ PinnedDispatcher, Dispatchers }
import scala.concurrent.Await
import akka.pattern.ask

object PinnedActorSpec {
  val config = """
    pinned-dispatcher {
      executor = thread-pool-executor
      type = PinnedDispatcher
    }
    """

  class TestActor extends Actor {
    def receive = {
      case "Hello"   ? sender() ! "World"
      case "Failure" ? throw new RuntimeException("Expected exception; to test fault-tolerance")
    }
  }
}

@org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner])
class PinnedActorSpec extends AkkaSpec(PinnedActorSpec.config) with BeforeAndAfterEach with DefaultTimeout {
  import PinnedActorSpec._

  private val unit = TimeUnit.MILLISECONDS

  "A PinnedActor" must {

    "support tell" in {
      var oneWay = new CountDownLatch(1)
      val actor = system.actorOf(Props(new Actor { def receive = { case "OneWay" ? oneWay.countDown() } }).withDispatcher("pinned-dispatcher"))
      val result = actor ! "OneWay"
      assert(oneWay.await(1, TimeUnit.SECONDS))
      system.stop(actor)
    }

    "support ask/reply" in {
      val actor = system.actorOf(Props[TestActor].withDispatcher("pinned-dispatcher"))
      assert("World" === Await.result(actor ? "Hello", timeout.duration))
      system.stop(actor)
    }
  }
} 

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

import java.util.UUID

import akka.actor.{Actor, Props}
import pl.mtomanski.fsmdemo.actors.ConnectionActor._
import pl.mtomanski.fsmdemo.domain.{Connection, Destination, Origin}

class ConnectionActor extends Actor {

  override def receive = {
    case FetchSoonestConnections(origin) =>
      println("Connection actor is fetching soonest connections")
      sender() ! getSoonestConnections(origin)
  }

  private def getSoonestConnections(origin: Origin) = {
    val soonestConnections = Seq(
      Connection("1", origin, destination1, departure1),
      Connection("2", origin, destination2, departure2)
    )
    SoonestConnectionsFromOrigin(soonestConnections)
  }
}

object ConnectionActor {

  def props(): Props = Props(new ConnectionActor)

  case class FetchSoonestConnections(origin: Origin)

  case class SoonestConnectionsFromOrigin(connections: Seq[Connection])

  // Mocked
  val destination1 = Destination(UUID.randomUUID().toString, "Wroclaw")
  val destination2 = Destination(UUID.randomUUID().toString, "Warsaw")
  val departure1 = "18:15"
  val departure2 = "18:30"
} 

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

import java.net.URL

import akka.actor.Status.Failure
import akka.actor.{ Actor, Props }
import akka.pattern.pipe
import mesosphere.marathon.ResolveArtifactsCanceledException
import mesosphere.marathon.io.storage.StorageProvider
import mesosphere.marathon.io.{ CancelableDownload, PathFun }
import mesosphere.util.Logging

import scala.concurrent.Promise

private[this] class ResolveArtifactsActor(
  url2Path: Map[URL, String],
  promise: Promise[Boolean],
  storage: StorageProvider)
    extends Actor
    with PathFun
    with Logging {

  import mesosphere.marathon.upgrade.ResolveArtifactsActor.DownloadFinished

  // all downloads that have to be performed by this actor
  var downloads = url2Path.map { case (url, path) => new CancelableDownload(url, storage, path) }

  override def preStart(): Unit = {
    import context.dispatcher
    downloads.map(_.get.map(DownloadFinished) pipeTo self)
    if (url2Path.isEmpty) promise.success(true) // handle empty list
  }

  override def postStop(): Unit = {
    downloads.foreach(_.cancel()) // clean up not finished artifacts
  }

  override def receive: Receive = {
    case DownloadFinished(download) =>
      downloads = downloads.filter(_ != download)
      if (downloads.isEmpty) promise.success(true)
    case Failure(ex) =>
      log.warn("Can not resolve artifact", ex) // do not fail the promise!
    case DeploymentActor.Shutdown =>
      if (!promise.isCompleted)
        promise.tryFailure(new ResolveArtifactsCanceledException("Artifact Resolving has been cancelled"))
      context.stop(self)
  }
}

object ResolveArtifactsActor {
  def props(
    url2Path: Map[URL, String],
    promise: Promise[Boolean],
    storage: StorageProvider): Props = Props(new ResolveArtifactsActor(url2Path, promise, storage))

  case class DownloadFinished(download: CancelableDownload)
} 

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

import com.franklevering.ports.adapters.http.HttpAdapter
import akka.actor.{ActorSystem, Props}
import akka.stream.ActorMaterializer
import com.franklevering.banking.domain.model.account.AccountRepository
import com.franklevering.banking.infrastructure.dependencyinjection.extension.GuiceAkkaExtension
import com.franklevering.banking.infrastructure.dependencyinjection.modules.{AkkaModule, InfrastructureModule, JournalModule}
import com.franklevering.ports.adapters.journal.{CassandraJournal, EngageStreaming}
import com.google.inject.Guice
import net.codingwell.scalaguice.InjectorExtensions._

object Application {
  def main(args: Array[String]): Unit = {
    val injector = Guice.createInjector(
      new AkkaModule(),
      new InfrastructureModule(),
      new JournalModule()
    )

    implicit  val system = injector.instance[ActorSystem]
    implicit val materializer = ActorMaterializer()
    implicit val executionContext = system.dispatcher
    system.actorOf(Props(classOf[CassandraJournal], injector.instance[AccountRepository])) ! EngageStreaming()
    HttpAdapter.startServer
  }
} 

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

import scala.reflect.ClassTag

import akka.actor.SupervisorStrategy.Stop
import akka.actor.{Actor, OneForOneStrategy, Props, Terminated}
import fs2.Task
import fs2.async.mutable.{Queue, Signal}

private class ClientActor[Out](props: Props, outQueue: Queue[Task, Out], closeSignal: Signal[Task, Boolean])
                              (implicit messageType: ClassTag[Out]) extends Actor {
  val serverActor = context actorOf props
  context watch serverActor
  
  def receive: Receive = {
    case Terminated(`serverActor`) =>
      closeSignal.set(true).unsafeRun()
      context stop self
    case messageType(m) if sender() == serverActor =>
      outQueue.enqueue1(m).unsafeRun()
    case m if sender() == serverActor =>
      org.log4s.getLogger.error(s"Server sent unhandled message ${m.getClass.getSimpleName} " +
        s"expecting a ${messageType.runtimeClass.getSimpleName}!")
    case m if sender() == self =>
      serverActor ! m
  }
  
  override def supervisorStrategy: OneForOneStrategy = OneForOneStrategy() {
    case _ => Stop
  }
} 

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

import akka.actor.{ActorSystem, Props}
import common.Tool._
import db.BackList


object SDKMain {
  def main(args: Array[String]) {
    UserApi.updateUsers
    run(Task.runTasks())
    //??????Actor
    val system = ActorSystem.create("PaiPai")
    val userActor = system.actorOf(Props[UserActor], name = "user_actor")
    val loanActor = system.actorOf(Props(new LoanActor(userActor)), name = "loan_actor")
    LoansApi.loanActor(loanActor)
  }

} 

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

import akka.testkit.AkkaSpec
import akka.actor.Props
import akka.actor.Actor
import akka.actor.ActorRef
import akka.actor.LocalActorRef
import scala.concurrent.duration._
import akka.actor.Identify
import akka.actor.ActorIdentity

@org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner])
class RouteeCreationSpec extends AkkaSpec {

  "Creating Routees" must {

    "result in visible routees" in {
      val N = 100
      system.actorOf(RoundRobinPool(N).props(Props(new Actor {
        system.actorSelection(self.path).tell(Identify(self.path), testActor)
        def receive = Actor.emptyBehavior
      })))
      for (i ? 1 to N) {
        expectMsgType[ActorIdentity] match {
          case ActorIdentity(_, Some(_)) ? // fine
          case x                         ? fail(s"routee $i was not found $x")
        }
      }
    }

    "allow sending to context.parent" in {
      val N = 100
      system.actorOf(RoundRobinPool(N).props(Props(new Actor {
        context.parent ! "one"
        def receive = {
          case "one" ? testActor forward "two"
        }
      })))
      val gotit = receiveWhile(messages = N) {
        case "two" ? lastSender.toString
      }
      expectNoMsg(100.millis)
      if (gotit.size != N) {
        fail(s"got only ${gotit.size} from [${gotit mkString ", "}]")
      }
    }

  }

} 

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

import scala.concurrent.duration._
import akka.actor.Actor
import akka.actor.ActorRef
import akka.actor.ActorSystem
import akka.actor.Address
import akka.actor.Props
import akka.testkit._

object AddressTerminatedTopicBenchSpec {

  class Subscriber(testActor: ActorRef) extends Actor {
    AddressTerminatedTopic(context.system).subscribe(self)
    testActor ! "started"

    override def postStop(): Unit = {
      AddressTerminatedTopic(context.system).unsubscribe(self)
    }

    def receive = Actor.emptyBehavior
  }
}

@org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner])
class AddressTerminatedTopicBenchSpec extends AkkaSpec("akka.loglevel=INFO") {
  import AddressTerminatedTopicBenchSpec._

  "Subscribe and unsubscribe of AddressTerminated" must {

    "be quick" in {
      val sys = ActorSystem(system.name + "2", system.settings.config)
      try {
        val num = 20000

        val t1 = System.nanoTime()
        val p = Props(classOf[Subscriber], testActor)
        val subscribers = Vector.fill(num)(sys.actorOf(p))
        receiveN(num, 10.seconds)
        log.info("Starting {} actors took {} ms", num, (System.nanoTime() - t1).nanos.toMillis)

        val t2 = System.nanoTime()
        shutdown(sys, 10.seconds, verifySystemShutdown = true)
        log.info("Stopping {} actors took {} ms", num, (System.nanoTime() - t2).nanos.toMillis)
      } finally {
        if (!sys.isTerminated) shutdown(sys)
      }
    }

  }
} 

//设置package包名称以及导入依赖的类
package sample.remote.calculator

import scala.concurrent.duration._
import scala.util.Random
import com.typesafe.config.ConfigFactory
import akka.actor.ActorSystem
import akka.actor.Props

object LookupApplication {
  def main(args: Array[String]): Unit = {
    if (args.isEmpty || args.head == "Calculator")
      startRemoteCalculatorSystem()
    if (args.isEmpty || args.head == "Lookup")
      startRemoteLookupSystem()
  }

  def startRemoteCalculatorSystem(): Unit = {
    val system = ActorSystem("CalculatorSystem",
      ConfigFactory.load("calculator"))
    system.actorOf(Props[CalculatorActor], "calculator")

    println("Started CalculatorSystem - waiting for messages")
  }

  def startRemoteLookupSystem(): Unit = {
    val system =
      ActorSystem("LookupSystem", ConfigFactory.load("remotelookup"))
    val remotePath =
      "akka.tcp://[email protected]:2552/user/calculator"
    val actor = system.actorOf(Props(classOf[LookupActor], remotePath), "lookupActor")

    println("Started LookupSystem")
    import system.dispatcher
    system.scheduler.schedule(1.second, 1.second) {
      if (Random.nextInt(100) % 2 == 0)
        actor ! Add(Random.nextInt(100), Random.nextInt(100))
      else
        actor ! Subtract(Random.nextInt(100), Random.nextInt(100))
    }
  }
} 

//设置package包名称以及导入依赖的类
package sample.remote.calculator

import scala.concurrent.duration._
import com.typesafe.config.ConfigFactory
import scala.util.Random
import akka.actor.ActorSystem
import akka.actor.Props

object CreationApplication {
  def main(args: Array[String]): Unit = {
    if (args.isEmpty || args.head == "CalculatorWorker")
      startRemoteWorkerSystem()
    if (args.isEmpty || args.head == "Creation")
      startRemoteCreationSystem()
  }

  def startRemoteWorkerSystem(): Unit = {
    ActorSystem("CalculatorWorkerSystem", ConfigFactory.load("calculator"))
    println("Started CalculatorWorkerSystem")
  }

  def startRemoteCreationSystem(): Unit = {
    val system =
      ActorSystem("CreationSystem", ConfigFactory.load("remotecreation"))
    val actor = system.actorOf(Props[CreationActor],
      name = "creationActor")

    println("Started CreationSystem")
    import system.dispatcher
    system.scheduler.schedule(1.second, 1.second) {
      if (Random.nextInt(100) % 2 == 0)
        actor ! Multiply(Random.nextInt(20), Random.nextInt(20))
      else
        actor ! Divide(Random.nextInt(10000), (Random.nextInt(99) + 1))
    }

  }
} 

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

import akka.actor.Actor
import akka.actor.Props

class HelloWorld extends Actor {

  override def preStart(): Unit = {
    // create the greeter actor
    val greeter = context.actorOf(Props[Greeter], "greeter")
    // tell it to perform the greeting
    greeter ! Greeter.Greet
  }

  def receive = {
    // when the greeter is done, stop this actor and with it the application
    case Greeter.Done => context.stop(self)
  }
} 

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

import scala.concurrent.duration._
import akka.testkit.AkkaSpec
import akka.actor.Actor
import akka.actor.ActorLogging
import akka.actor.Props
import akka.event.Logging
import akka.testkit.ImplicitSender
import akka.testkit.TestProbe
import akka.actor.Deploy
import akka.event.Logging.Info
import akka.actor.ExtendedActorSystem

object LogSourceSpec {
  class Reporter extends Actor with ActorLogging {
    def receive = {
      case s: String ?
        log.info(s)
    }
  }
}

@org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner])
class LogSourceSpec extends AkkaSpec(
  """
    akka.loglevel = INFO
    akka.actor.provider = "akka.remote.RemoteActorRefProvider"
    akka.remote.netty.tcp.port = 0
  """) {

  import LogSourceSpec._

  val reporter = system.actorOf(Props[Reporter], "reporter")
  val logProbe = TestProbe()
  system.eventStream.subscribe(system.actorOf(Props(new Actor {
    def receive = {
      case i @ Info(_, _, msg: String) if msg contains "hello" ? logProbe.ref ! i
      case _ ?
    }
  }).withDeploy(Deploy.local), "logSniffer"), classOf[Logging.Info])

  "Log events" must {

    "should include host and port for local LogSource" in {
      reporter ! "hello"
      val info = logProbe.expectMsgType[Info]
      info.message should be("hello")
      val defaultAddress = system.asInstanceOf[ExtendedActorSystem].provider.getDefaultAddress
      info.logSource should include(defaultAddress.toString)
    }
  }
} 

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

//#plain-spec
import akka.actor.ActorSystem
import akka.actor.Actor
import akka.actor.Props
import akka.testkit.{ TestActors, TestKit, ImplicitSender }
import org.scalatest.WordSpecLike
import org.scalatest.Matchers
import org.scalatest.BeforeAndAfterAll

//#implicit-sender
class MySpec(_system: ActorSystem) extends TestKit(_system) with ImplicitSender
  with WordSpecLike with Matchers with BeforeAndAfterAll {
  //#implicit-sender

  def this() = this(ActorSystem("MySpec"))

  override def afterAll {
    TestKit.shutdownActorSystem(system)
  }

  "An Echo actor" must {

    "send back messages unchanged" in {
      val echo = system.actorOf(TestActors.echoActorProps)
      echo ! "hello world"
      expectMsg("hello world")
    }

  }
}
//#plain-spec 

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

import akka.actor.{ActorRef, ActorSystem, Props}
import akka.cluster.sharding.{ClusterSharding, ClusterShardingSettings, ShardRegion}
import lib.MormontConfig
import lib.cluster.MormontShardRegion.MormontShardRegionActorRef
import lib.models.PayLoad


abstract class MormontShardRegion[T <: PayLoad](actorSystem: ActorSystem) {
  private val prefix : String = "shard-region-"
  def channelName: String

  def entityProps: Props

  final private val extractEntityId: ShardRegion.ExtractEntityId = {
    case p: PayLoad ?
      (p.persistenceId, p)
  }

  final private val extractShardId: ShardRegion.ExtractShardId = {
    case p: PayLoad ?
      (math.abs(p.persistenceId.hashCode) % MormontConfig.numOfShards).toString
  }

  final def startShard(): MormontShardRegionActorRef = {
   ClusterSharding(actorSystem).start(
      typeName = prefix + channelName,
      entityProps = entityProps,
      settings = ClusterShardingSettings(actorSystem),
      extractEntityId = extractEntityId,
      extractShardId = extractShardId
    )
  }
}

object MormontShardRegion {
  type MormontShardRegionActorRef = ActorRef
} 

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

import akka.actor.{ActorRef, ActorSystem, Props}
import akka.cluster.singleton.{ClusterSingletonManager, ClusterSingletonManagerSettings}
import lib.models.DefaultCommands.Stop


abstract class MormontListenerSingleton(system: ActorSystem) {
  private val prefix: String = "listener-singleton-"

  protected def singletonProps: Props

  protected def channelName: String

  final def startSingleton(): Unit = {
    system.actorOf(
      ClusterSingletonManager.props(
        singletonProps = singletonProps,
        terminationMessage = Stop,
        settings = ClusterSingletonManagerSettings(system)),
      name = prefix + channelName)
  }
}

object MormontListenerSingleton {
  type MormontListenerSingletonActorRef = ActorRef
} 

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

import akka.actor.Props
import example.models.Command.GetVersion
import example.models.ExamplePayload
import lib.tools.Monitor


class ExampleMonitor extends Monitor {
  var versionState: String = "default version"
  var description: String = "default description"

  def updateState(payLoad: ExamplePayload): Unit = {
    versionState = payLoad.version
    description = payLoad.description
  }

  def validatePayload(payload: ExamplePayload): Boolean = {
    payload.name.nonEmpty && payload.description.nonEmpty && payload.version.nonEmpty
  }

  override def commandHandler = {
    case payload: ExamplePayload =>
      log.info(s"payload received: [$payload]")
      if (validatePayload(payload)) {
        persist(payload)(updateState)
      } else {
        log.error(s"Invalid payload: [$payload]")
        // Take necessary steps for the in-valid entry in Postgres, more
        // features for informing concerned users will be added in future like publishing
        // failures to a UI or a feature to send email.
      }
    case GetVersion => sender ! versionState
  }

  override def recoverHandler = {
    case payload: ExamplePayload => updateState(payload)
  }
}

object ExampleMonitor {
  def props(): Props = Props(new ExampleMonitor)
} 

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

import akka.actor.{Actor, Props}
import pl.mtomanski.fsmdemo.actors.ReservationActor.{CancelReservation, MakeReservation}
import pl.mtomanski.fsmdemo.domain.Connection


class ReservationActor extends Actor {
  override def receive: Receive = {
    case MakeReservation(connection) => println(s"Reservation made for connection ${connection.id}")
    case CancelReservation(connection) => println(s"Reservation CANCELLED for connection ${connection.id}")
  }
}

object ReservationActor {
  def props(): Props = Props(new ReservationActor)

  case class MakeReservation(connection: Connection)
  case class CancelReservation(connection: Connection)
} 

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



import akka.actor.{ActorSystem, Props}

object Main {
  val urls = List("http://www.amazon.com", "https://www.google.com", "http://www.cnn.com", "http://www.twitter.com")

  def main(args: Array[String]): Unit = {
    fetchSizesSequentially()
    fetchSizesConcurrently()
  }

  def fetchSizesSequentially() = {
    val start = System.nanoTime
    for (url <- urls) {
      println("Size for " + url + ": " + Scraper.getPageSize(url))
    }
    val end = System.nanoTime
    println("Fetching sequentially took " + (end - start) / 1000000000.0 + " seconds.")
  }

  def fetchSizesConcurrently() = {
    val system = ActorSystem("Main")
    for (url <- urls) {
      val ac = system.actorOf(Props(new PageSizeFetcher(url)))
    }
  }
} 

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

import akka.actor.{Actor, Props}

class PageSizeFetcher(url: String) extends Actor {

  override def preStart(): Unit = {
    val fetcher = context.actorOf(Props(new SizerActor()))
    fetcher ! SizerState.GetPageSize(url)
  }

  def receive = {
    case SizerState.Done(time: Long, amtOfLinks: Int) =>
      println("Fetching " + url + " took " + time / 1000000000.0 + " seconds. Had " + amtOfLinks + " links.")

  }

  class SizerActor() extends Actor {
    def receive = {
      case SizerState.GetPageSize(url: String) =>
        val start = System.nanoTime
        val list = scala.collection.mutable.ListBuffer[String]()
        for (line <- Scraper.getPage(url).getLines()) {
          val pattern = "(?i)<a.+?href=\\\"(http.+?)\\\".*?>(.+?)</a>".r
          for (m <- pattern.findAllIn(line)) {
            list += m
          }
        }
        sender ! SizerState.Done(System.nanoTime - start, list.size)
    }
  }

  object SizerState {

    case class GetPageSize(url: String)

    case class Done(time: Long, amtOfLinks: Int)

  }

}