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


import akka.actor.Actor
import akka.actor.ActorRef
import akka.actor.ActorLogging
import akka.actor.Props

import play.api.libs.json.Json.toJson

import bmmessages._
import akka.actor.Terminated

class RoutesActor extends Actor with ActorLogging {
	var originSender : ActorRef = null
	var next : ActorRef = null
	def receive = {
		case excute(msr)  => {
		    implicit val cm = msr.cm
		    
			originSender = sender
			msr.lst match {
				case Nil => originSender ! toJson("error")
				case head :: tail => {
					head match {
						case p : ParallelMessage => {
							next = context.actorOf(ScatterGatherActor.prop(self, MessageRoutes(tail, msr.rst)), "gate")
							next ! head
						}
						case c : CommonMessage => {
							next = context.actorOf(PipeFilterActor.prop(self, MessageRoutes(tail, msr.rst)), "gate")
							next ! head
						}
					}
					
					context.watch(next)
				}
			}
		}
		case result(rst) => {
			originSender ! rst
			cancelActor
		}
		case error(err) => {
			originSender ! err
			cancelActor
		}
		case bmmessages.timeout() => {
			originSender ! toJson("timeout")
			cancelActor
		}
		case Terminated(actorRef) => println("Actor {} terminated", actorRef)
		case _ => Unit
 	}
	
	def cancelActor = {
		context.stop(self)
	}
} 

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

import akka.actor.Actor
import akka.actor.ActorLogging
import akka.event.LoggingReceive
import akka.actor.ActorRef
import akka.actor.Terminated
import play.libs.Akka
import akka.actor.Props

class BoardActor extends Actor with ActorLogging {
  var users = Set[ActorRef]()

  def receive = LoggingReceive {
    case m:Message => users map { _ ! m}
    case Subscribe => {
      users += sender
      context watch sender
    }
    case Terminated(user) => users -= user
  }
}

object BoardActor {
  lazy val board = Akka.system().actorOf(Props[BoardActor])
  def apply() = board
}

case class Message(uuid: String, s: String)
object Subscribe 

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

import akka.actor.{ActorSystem, Terminated}
import game_of_life.actor.GridActor
import game_of_life.actor.GridActor.{Pause, Play}
import game_of_life.controller.GameController._
import game_of_life.model.Game

import scala.concurrent.Future


class GameController(game: Game)(view: Game#Grid => Unit = printGrid) {

  require(view != null)

  private val system = ActorSystem("game-of-life")

  private val grid = system.actorOf(GridActor.props(game)(view), "grid")

  private var playing = false

  def play(): Unit = {
    grid ! Play
    playing = true
  }

  def pause(): Unit = {
    grid ! Pause
    playing = false
  }

  def toggle(): Unit = {
    if (playing) {
      pause()
    } else {
      play()
    }
  }

  def stop(): Future[Terminated] = system terminate ()
}

object GameController {

  def printGrid(grid: Game#Grid): Unit = {
    import Console._
    print(grid)
    flush()
  }
} 

//设置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 sample.remote.calculator

import scala.concurrent.duration._
import akka.actor.Actor
import akka.actor.ActorIdentity
import akka.actor.ActorRef
import akka.actor.Identify
import akka.actor.ReceiveTimeout
import akka.actor.Terminated

class LookupActor(path: String) extends Actor {

  sendIdentifyRequest()

  def sendIdentifyRequest(): Unit = {
    context.actorSelection(path) ! Identify(path)
    import context.dispatcher
    context.system.scheduler.scheduleOnce(3.seconds, self, ReceiveTimeout)
  }

  def receive = identifying

  def identifying: Actor.Receive = {
    case ActorIdentity(`path`, Some(actor)) =>
      context.watch(actor)
      context.become(active(actor))
    case ActorIdentity(`path`, None) => println(s"Remote actor not available: $path")
    case ReceiveTimeout              => sendIdentifyRequest()
    case _                           => println("Not ready yet")
  }

  def active(actor: ActorRef): Actor.Receive = {
    case op: MathOp => actor ! op
    case result: MathResult => result match {
      case AddResult(n1, n2, r) =>
        printf("Add result: %d + %d = %d\n", n1, n2, r)
      case SubtractResult(n1, n2, r) =>
        printf("Sub result: %d - %d = %d\n", n1, n2, r)
    }
    case Terminated(`actor`) =>
      println("Calculator terminated")
      sendIdentifyRequest()
      context.become(identifying)
    case ReceiveTimeout =>
    // ignore

  }
} 

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

import akka.actor.ActorSystem
import akka.actor.Props
import akka.actor.ActorRef
import akka.actor.Actor
import akka.actor.ActorLogging
import akka.actor.Terminated

object Main2 {

  def main(args: Array[String]): Unit = {
    val system = ActorSystem("Hello")
    val a = system.actorOf(Props[HelloWorld], "helloWorld")
    system.actorOf(Props(classOf[Terminator], a), "terminator")
  }

  class Terminator(ref: ActorRef) extends Actor with ActorLogging {
    context watch ref
    def receive = {
      case Terminated(_) =>
        log.info("{} has terminated, shutting down system", ref.path)
        context.system.shutdown()
    }
  }

} 

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

import akka.actor.InternalActorRef
import akka.actor.Terminated
import akka.actor.Actor
import akka.actor.ActorRef
import akka.dispatch.sysmsg.DeathWatchNotification
import akka.dispatch.{ UnboundedMessageQueueSemantics, RequiresMessageQueue }


private[akka] class RemoteDeploymentWatcher extends Actor with RequiresMessageQueue[UnboundedMessageQueueSemantics] {
  import RemoteDeploymentWatcher._
  var supervisors = Map.empty[ActorRef, InternalActorRef]

  def receive = {
    case WatchRemote(a, supervisor: InternalActorRef) ?
      supervisors += (a -> supervisor)
      context.watch(a)

    case t @ Terminated(a) if supervisors isDefinedAt a ?
      // send extra DeathWatchNotification to the supervisor so that it will remove the child
      supervisors(a).sendSystemMessage(DeathWatchNotification(a, existenceConfirmed = t.existenceConfirmed,
        addressTerminated = t.addressTerminated))
      supervisors -= a

    case _: Terminated ?
  }
} 

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

import akka.actor.ActorSystem
import akka.actor.ExtendedActorSystem
import akka.actor.Actor
import akka.actor.Terminated
import akka.actor.ActorLogging
import akka.actor.Props
import akka.actor.ActorRef
import scala.util.control.NonFatal


  def main(args: Array[String]): Unit = {
    if (args.length != 1) {
      println("you need to provide exactly one argument: the class of the application supervisor actor")
    } else {
      val system = ActorSystem("Main")
      try {
        val appClass = system.asInstanceOf[ExtendedActorSystem].dynamicAccess.getClassFor[Actor](args(0)).get
        val app = system.actorOf(Props(appClass), "app")
        val terminator = system.actorOf(Props(classOf[Terminator], app), "app-terminator")
      } catch {
        case NonFatal(e) ? system.shutdown(); throw e
      }
    }
  }

  class Terminator(app: ActorRef) extends Actor with ActorLogging {
    context watch app
    def receive = {
      case Terminated(_) ?
        log.info("application supervisor has terminated, shutting down")
        context.system.shutdown()
    }
  }

} 

//设置package包名称以及导入依赖的类
package net.ruippeixotog.scalafbp.runtime

import akka.actor.{ Actor, Props, Terminated }

import net.ruippeixotog.scalafbp.component.Component
import net.ruippeixotog.scalafbp.component.ComponentActor._
import net.ruippeixotog.scalafbp.runtime.NetworkBroker.External

case class GraphComponent(graph: Graph) extends Component {
  val name = GraphComponent.getId(graph)
  val description = s"Executes the graph ${graph.id} as a component"
  val icon = Some("sitemap")

  lazy val inPorts = graph.publicIn.flatMap {
    case (public, PublicPort(ref, _)) => graph.inPortInfo(ref).map(_.withId(public))
  }.toList

  lazy val outPorts = graph.publicOut.flatMap {
    case (public, PublicPort(ref, _)) => graph.outPortInfo(ref).map(_.withId(public))
  }.toList

  val instanceProps = Props(new Actor {
    val brokerActor = context.actorOf(NetworkBroker.props(graph, false, self, self))
    context.watch(brokerActor)

    def receive = {
      case msg: ComponentMessage => brokerActor ! External(msg)
      case External(msg) => context.parent ! msg

      case msg: NetworkBroker.NodeCommand => context.parent ! msg.cmd
      case msg: NetworkBroker.NodeError => throw new Exception(msg.msg)
      case msg: NetworkBroker.NetworkError => throw new Exception(msg.msg)

      case Terminated(`brokerActor`) => context.stop(self)
    }
  })
}

object GraphComponent {
  def getId(graph: Graph) = "scalafbp/" + graph.id
} 

//设置package包名称以及导入依赖的类
package net.ruippeixotog.scalafbp.runtime

import scala.language.reflectiveCalls

import akka.actor.Terminated
import spray.json.DefaultJsonProtocol._

import net.ruippeixotog.scalafbp.component.ComponentActor.Incoming

class NetworkBrokerInitialsSpec extends NetworkBrokerSpec {

  "A NetworkBroker" should {

    "handle initial values correctly" in {

      "forward initial values into components as Incoming messages" in new BrokerInstance {
        def _graph = graph

        lazy val graph = new SingleNodeGraph {
          val n1Probe = probeBehavior(n1)
          initial("aaa") ~> (n1, 1)
        }

        graph.n1Probe must receive(Incoming("in1", "aaa"))
      }

      "fail if an initial value cannot be converted to the target type" in new BrokerInstance {
        def _graph = graph

        lazy val graph = new SingleNodeGraph {
          val n1Probe = probeBehavior(n1)
          initial(4) ~> (n1, 1)
        }

        lifeProbe must receive.like { case Terminated(`broker`) => ok }
        outputProbe must receive.like {
          case NetworkBroker.NetworkError(msg) => msg mustEqual "Could not deserialize initial data for n1[in1]"
        }.afterOthers
      }
    }
  }
} 

//设置package包名称以及导入依赖的类
package net.ruippeixotog.scalafbp.runtime

import scala.language.reflectiveCalls

import akka.actor.{ Actor, Props, Terminated }
import akka.testkit.TestProbe

class NetworkBrokerLifetimeSpec extends NetworkBrokerSpec {

  "A NetworkBroker" should {

    "manage the lifetime of the components" in {

      "instantiate all components in a graph when the network starts" in new BrokerInstance {
        def _graph = graph

        lazy val probe = TestProbe()
        def instanceProps(id: String) = Props(new Actor {
          probe.ref ! s"started_$id"
          def receive = Actor.ignoringBehavior
        })

        lazy val graph = new TwoNodeGraph {
          behavior(n1, instanceProps("n1"))
          behavior(n2, instanceProps("n2"))
        }

        probe must receive.allOf("started_n1", "started_n2")
      }

      "terminate the network when all components terminate" in new BrokerInstance {
        def _graph = graph

        lazy val instanceProps = Props(new Actor {
          context.stop(self)
          def receive = Actor.ignoringBehavior
        })

        lazy val graph = new TwoNodeGraph {
          behavior(n1, instanceProps)
          behavior(n2, instanceProps)
        }

        lifeProbe must receive.like { case Terminated(`broker`) => ok }
      }
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.bob.scalatour.akka.cluster

import akka.actor.Terminated
import akka.cluster.ClusterEvent._

class EventCollector extends ClusterRoledWorker {

  @volatile var recordCounter: Int = 0

  override def receive: Receive = {
    case MemberUp(member) => println(s"Member is Up: ${member.address}")
    case UnreachableMember(member) => println(s"Member detected as Unreachable: ${member}")
    case MemberRemoved(member, previousStatus) => println(s"Member is removed: ${member.address} after ${previousStatus}")
    case Registration => {
      context.watch(sender)
      workers = workers :+ sender
      println(s"Interceptor registered: ${sender}, Registered interceptors: ${workers.size}")
    }
    case Terminated(interceptingActorRef) => workers = workers.filterNot(_ == interceptingActorRef)
    case RawNginxRecord(sourceHost, line) => {
      val eventCode = "eventcode=(\\d+)".r.findFirstIn(line).getOrElse("")
      println(s"Raw message: eventCode=${eventCode}, sourceHost=${sourceHost}, line=${line}")
      recordCounter += 1
      if (workers.size > 0) {
        // ??Roudrobin?????????????????interceptor????
        val interceptorIndex = (if (recordCounter < 0) 0 else recordCounter) % workers.size
        workers(interceptorIndex) ! NginxRecord(sourceHost, eventCode, line)
        println("Details: interceptorIndex=" + interceptorIndex + ", interceptors=" + workers.size)
      }
    }
    case _: MemberEvent =>
  }

} 

//设置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 wow.realm.session

import akka.actor.{Actor, ActorLogging, ActorRef, Props, Terminated}
import wow.auth.AccountsState
import wow.auth.AccountsState.NotifyAccountOnline
import wow.realm.entities.Guid
import wow.realm.protocol._
import wow.realm.session.Session.CreatePlayer
import wow.realm.{RealmContext, RealmContextData}


class Session(val login: String, override val networkWorker: ActorRef)(override implicit val realm: RealmContextData)
  extends Actor
          with ActorLogging
          with RealmContext
          with PacketHandlerTag
          with ForwardToNetworkWorker {
  var player: Option[ActorRef] = _

  context.actorSelection(AccountsState.ActorPath) ! NotifyAccountOnline(login, networkWorker)

  override def receive: Receive = {
    case CreatePlayer(guid: Guid) =>
      val ref = context.actorOf(SessionPlayer.props(guid, networkWorker), SessionPlayer.PreferredName(guid))
      player = Some(ref)
      sender() ! ref
      context.watch(ref)

    case NetworkWorker.HandlePacket(header, payloadBits) =>
      PacketHandler(header, payloadBits)(this)

    case Terminated(subject) if subject == player.getOrElse(ActorRef.noSender) =>
      // Bring player back to character screen is its player actor crashes
      sendOpCode(OpCodes.SLogoutComplete)
  }
}

object Session {
  def props(login: String, networkWorker: ActorRef)(implicit realm: RealmContextData): Props =
    Props(new Session(login, networkWorker))

  def PreferredName(login: String) = s"session-$login"

  case class CreatePlayer(guid: Guid)

} 

//设置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 com.udemy.akka

import akka.actor.{Actor, ActorRef, ActorSystem, Props, Terminated}
import akka.actor.Actor.Receive




class Ares(athena:ActorRef) extends Actor{
  override def preStart(): Unit = {
    context.watch(athena)
  }

  override def postStop(): Unit = {
    println("Ares post stop")
  }
  override def receive: Receive = {
    case Terminated=>{

      context.stop(self)
    }
  }
}
class Athena extends Actor{
  override def receive: Receive = {
    case msg=>
      println(s"Athena recieved ${msg}")
      context.stop(self)
  }
}


object Monitor extends App{
  val system =ActorSystem("MonitorSystem")


  private val athena: ActorRef = system.actorOf(Props[Athena],"athena")
  private val ares: ActorRef = system.actorOf(Props(new Ares(athena)),"ares")
  athena ! "hello"
  system.terminate()
} 

//设置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.upgrade

import akka.actor.{ ActorLogging, Terminated, Actor, ActorRef }
import mesosphere.marathon.upgrade.DeploymentActor.Cancel
import scala.concurrent.Promise

class StopActor(toStop: ActorRef, promise: Promise[Boolean], reason: Throwable) extends Actor with ActorLogging {

  override def preStart(): Unit = {
    context.watch(toStop)
    toStop ! Cancel(reason)
  }

  def receive: Receive = {
    case Terminated(`toStop`) =>
      promise.success(true)
      log.debug(s"$toStop has successfully been stopped.")
      context.unwatch(toStop)
      context.stop(self)
  }
} 

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

import akka.actor.ActorDSL._
import akka.actor.{ ActorSystem, PoisonPill, Terminated }
import akka.event.EventStream
import akka.testkit.TestProbe
import mesosphere.marathon.core.event.MarathonEvent

import scala.collection.immutable.Seq
import scala.concurrent.Await
import scala.concurrent.duration.Duration

class CaptureEvents(eventStream: EventStream) {
  
  def forBlock(block: => Unit): Seq[MarathonEvent] = {
    implicit val actorSystem = ActorSystem("captureEvents")

    // yes, this is ugly. Since we only access it in the actor until it terminates, we do have
    // the correct thread sync boundaries in place.

    var capture = Vector.empty[MarathonEvent]
    val captureEventsActor = actor {
      new Act {
        become {
          case captureMe: MarathonEvent => capture :+= captureMe
        }
      }
    }
    eventStream.subscribe(captureEventsActor, classOf[MarathonEvent])
    eventStream.subscribe(captureEventsActor, classOf[String])

    try {
      block
    } finally {
      eventStream.unsubscribe(captureEventsActor)
      captureEventsActor ! PoisonPill
      val probe = TestProbe()
      probe.watch(captureEventsActor)
      probe.expectMsgClass(classOf[Terminated])
      Await.result(actorSystem.terminate(), Duration.Inf)
    }

    capture
  }

} 

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

import akka.actor.{Actor, ActorLogging, ActorRef, ActorSystem, Props, Terminated}
import com.taintech.aff.actor.AmazonFineFoodManager

object AmazonFineFoodApp {

  def main(args: Array[String]): Unit = {
    val system = ActorSystem("amazon-fine-foods")
    val manager = system.actorOf(
      AmazonFineFoodManager.props(args.headOption.getOrElse("/Users/taintech/Downloads/amazon-fine-foods/Reviews.csv")), "manager"
    )
    system.actorOf(Props(classOf[Terminator], manager), "terminator")
  }

  class Terminator(ref: ActorRef) extends Actor with ActorLogging {
    context watch ref
    def receive = {
      case Terminated(_) =>
        log.info("{} has terminated, shutting down system", ref.path)
        context.system.terminate()
    }
  }

}