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

import akka.actor.Actor
import java.awt._
import javax.swing.JPanel
import javax.swing.SwingUtilities
import java.awt.image.BufferedImage


class BubblePanel extends JPanel {

  private var bubble: Bubble = null

  def doRedraw(bubble: Bubble) {
    this.bubble = bubble
    // TODO: i don't know if this is the "right" way to do this, but the normal
    // 'repaint' wasn't cutting it.
    val redrawHeight = bubble.circleDiameter * 2
    paintImmediately(bubble.x, bubble.y, bubble.circleDiameter, redrawHeight)
  }

  // the bubbles now draw themselves 
  override def paintComponent(g: Graphics) {
    if (bubble != null) {
      val component = this.asInstanceOf[Component]
      bubble.drawBubbleFast(g, component.getGraphicsConfiguration)
    }
  }

}

////////////////////////////////////////////////////////////////

case class Redraw(bubble: Bubble)

class BubblePanelActor(bubblePanel: BubblePanel) extends Actor {

  def receive = {
    case Redraw(bubble) => doRedraw(bubble) 
    case _ =>
  }

  private def doRedraw(bubble: Bubble) {
    SwingUtilities.invokeLater(new Runnable {
      def run {
        bubblePanel.doRedraw(bubble)
      }
    })
  }

} 

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

import akka.actor.{Actor, ActorRef}
import akka.event.LoggingReceive
import reactive4.homework.AuctionSearch.{AddAuction, SearchAuction, SearchResult}

class AuctionSearch extends Actor {

  var map:Map[String, ActorRef] = Map()

  override def receive: Receive = LoggingReceive {
    case msg: AddAuction =>
      map = map + ((msg.title, msg.auction))
    case msg: SearchAuction =>
      val list: List[ActorRef] = map.filterKeys(_.contains(msg.query)).values.toList
      sender() ! SearchResult(msg.query, list)
  }
}

object AuctionSearch {

  case class AddAuction(title: String, auction:ActorRef)
  case class SearchAuction(query: String)
  case class SearchResult(query: String, auctions: List[ActorRef])

} 

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

import akka.actor.{ActorLogging, PoisonPill, Actor, ReceiveTimeout}
import akka.cluster.sharding.ShardRegion.Passivate

trait Passivation extends ActorLogging {
  this: Actor =>

  protected def passivate(receive: Receive): Receive = receive.orElse {
    // tell parent actor to send us a poisinpill
    case ReceiveTimeout =>
      log.info(s" $self ReceiveTimeout: passivating. ")
      context.parent ! Passivate(stopMessage = PoisonPill)

    // stop
    case PoisonPill =>
      log.info(s" $self PoisonPill")
      context.stop(self)
  }
} 

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

import akka.actor.Actor
import java.io._
import sun.audio._
import javax.sound.sampled.AudioSystem

case object PlaySuccessSound
case object PlayFailureSound
    
class PlaySoundActor extends Actor {

  private val SUCCESS_SOUND_FILENAME = "Synth-Zingers-04.aif"
  private val FAILURE_SOUND_FILENAME = "Comedy-Low-Honk.aif"

  def receive = {
    case PlaySuccessSound => playSoundFile(SUCCESS_SOUND_FILENAME)
    case PlayFailureSound => playSoundFile(FAILURE_SOUND_FILENAME)
    case _ =>
  }

  def playSoundFile(filename: String) {
    val classloader = getClass.getClassLoader
    val inputStream = classloader.getResourceAsStream(filename)
    val audioStream = new AudioStream(inputStream)
    AudioPlayer.player.start(audioStream)
    
    // TODO sound won't play if the file is closed right away
    //inputStream.close
  }

} 

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

import akka.actor.Actor
import com.sksamuel.scrimage.Image
import com.sksamuel.scrimage.filter.{SparkleFilter => ScrimageSparkleFilter}
import iosr.Messages.{Response, SparkleCommand}

class SparkleFilter extends Actor {
  override def receive: Receive = {
    case SparkleCommand(imageBytes, params) =>
      val senderActor = sender()
      val image = Image(imageBytes)
      senderActor ! Response(
        image.filter(ScrimageSparkleFilter(
          rays = params.rays,
          radius = params.radius,
          amount = params.amount
        )).bytes
      )
  }
} 

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

import akka.actor.Actor
import com.sksamuel.scrimage.Image
import com.sksamuel.scrimage.filter.{ContrastFilter => ScrimageContrastFilter}
import iosr.Messages.{ContrastCommand, Response}

class ContrastFilter extends Actor {
  override def receive: Receive = {
    case ContrastCommand(imageBytes, params) =>
      val senderActor = sender()
      val image = Image(imageBytes)
      senderActor ! Response(
        image.filter(ScrimageContrastFilter(params.contrast)).bytes
      )
  }
} 

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

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

class HelloWorldActor extends Actor {

  override def preStart(): Unit = {
    // create the greeter actor
    val greeter = context.actorOf(Props[GreeterActor], "greeter")

    // Send it the 'Greet' message
    greeter ! GreeterMessages.Greet
  }

  def receive = {
    // When we receive the 'Done' message, stop this actor
    // (which if this is still the initialActor will trigger the deathwatch and stop the entire ActorSystem)
    case GreeterMessages.Done => {
      context.stop(self)
    }
  }
} 

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

import akka.actor.{ActorLogging, Actor}
import scala.collection.mutable


class SessionActor extends Actor with ActorLogging {

  val users = mutable.Map.empty[String, UserData]

  def incrementVictoryCount(userName: String, selfVictory: Boolean) = users.get(userName) match {
    case Some(userData @ UserData(selfCount, aiCount)) =>
      users.update(userName, if (selfVictory) UserData(selfCount + 1, aiCount) else UserData(selfCount, aiCount + 1))
      log.info(s"Cannot update victory count. User $userName does not exist")
    case _ => log.warning(s"Cannot update victory count. User $userName does not exist")
  }

  def receive = {
    case msg: SessionMessage => msg match {
      case SignIn(userName) =>
        if(!users.isDefinedAt(userName)) users + (userName -> UserData(0, 0))
        log.info(s"User $userName successfully signed in")
        sender() ! userName
      case SelfVictory(userName) => incrementVictoryCount(userName, selfVictory = true)
      case AIVictory(userName) => incrementVictoryCount(userName, selfVictory = false)
    }
    case msg => log.warning(s"unknown message: $msg")
  }
}

case class UserData(selfWin: Int, aiWin: Int)

sealed trait SessionMessage
case class SignIn(userName: String) extends SessionMessage
case class SelfVictory(userName: String) extends SessionMessage
case class AIVictory(userName: String) extends SessionMessage 

//设置package包名称以及导入依赖的类
package com.github.unknownnpc.remotedebugtool.actor

import akka.actor.{Actor, ActorLogging, ActorRef, Props}
import com.github.unknownnpc.remotedebugtool.config.{AppConfig, RemoteDebugToolConfig}
import com.github.unknownnpc.remotedebugtool.domain._
import com.github.unknownnpc.remotedebugtool.exception.ReportException
import com.github.unknownnpc.remotedebugtool.message.{MainAppActorStop, ReportServicePayload, ReportServicePrint}

import scala.collection.mutable.ListBuffer

class ReportServiceActor(mainAppActorRef: ActorRef) extends Actor with ActorLogging {

  self: AppConfig =>

  val values = ListBuffer.empty[ReportRow]

  override def receive = {

    case ReportServicePayload(payload) =>
      log.debug(s"Print service received incoming payload: [$payload]")
      values += reportRowFrom(payload)

    case ReportServicePrint =>
      log.debug(s"Received print command")
      log.info(systemConfig.reportFormatter.format(values.toList))
      mainAppActorRef ! MainAppActorStop

  }


  private def reportRowFrom(payload: BreakpointPayload) = {
    val testTarget = findServerById(payload.breakpoint.targetId)
    JvmReportRow(testTarget.id,
      testTarget.address,
      testTarget.port,
      payload.breakpoint.line,
      payload.breakpoint.className,
      payload.breakpointValue
    )
  }

  private def findServerById(id: ID) = {
    servers.find(_.id == id).getOrElse(
      throw ReportException("Unable to match payload to server instance")
    )
  }

}

object ReportServiceActor {
  def props(mainAppActorRef: ActorRef) =
    Props(new ReportServiceActor(mainAppActorRef) with RemoteDebugToolConfig)
} 

//设置package包名称以及导入依赖的类
package com.github.unknownnpc.remotedebugtool.actor

import akka.actor.{Actor, ActorLogging, ActorRef, ActorSystem}
import com.github.unknownnpc.remotedebugtool.message._

import scala.concurrent.ExecutionContext

class MainAppActor()(implicit actorSystem: ActorSystem) extends Actor with ActorLogging {

  private implicit val dispatcher: ExecutionContext = actorSystem.dispatcher

  private var jdiVmServiceActor: ActorRef = _
  private var reportServiceActor: ActorRef = _

  override def preStart(): Unit = {
    log.info("Main app actor starts")
    reportServiceActor = createReportServiceActor()
    jdiVmServiceActor = createJdiVmServiceActor()
  }

  override def postStop() {
    log.info("Main app actor stops")
  }

  override def receive: Receive = {

    case MainAppActorStart =>
      startServices()

    case MainAppActorStop =>
      stopServices()
      context.stop(self)
      context.system.terminate()

  }

  private def startServices() = {
    jdiVmServiceActor ! JdiVmServiceStart
  }

  private def stopServices() = {
    log.info("Received command to stop all services")
    jdiVmServiceActor ! JdiVmServiceStop
    reportServiceActor ! ReportServiceStop
  }

  private def createJdiVmServiceActor() = {
    context.actorOf(JdiVmServiceActor.props(reportServiceActor), "jdi-vm-service")
  }

  private def createReportServiceActor() = {
    context.actorOf(ReportServiceActor.props(self), "report-service")
  }

} 

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

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.routing._

class RouteeCreationSpec extends AkkaSpec {

  "Creating Routees" must {

    "result in visible routees" in {
      val N = 100
      system.actorOf(Props(new Actor {
        testActor ! system.actorFor(self.path)
        def receive = Actor.emptyBehavior
      }).withRouter(RoundRobinRouter(N)))
      for (i ? 1 to N) {
        expectMsgType[ActorRef] match {
          case _: LocalActorRef ? // fine
          case x                ? fail(s"routee $i was a ${x.getClass}")
        }
      }
    }

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

  }

} 

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

import akka.actor.Actor
import akka.actor.ActorSystem
import com.typesafe.config.ConfigFactory
import akka.actor.Props

object Receiver {
  def main(args: Array[String]): Unit = {
    val system = ActorSystem("Sys", ConfigFactory.load("remotelookup"))
    system.actorOf(Props[Receiver], "rcv")
  }
}

class Receiver extends Actor {
  import Sender._

  def receive = {
    case m: Echo  => sender() ! m
    case Shutdown => context.system.shutdown()
    case _        =>
  }
} 

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

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

class CreationActor extends Actor {

  def receive = {
    case op: MathOp =>
      val calculator = context.actorOf(Props[CalculatorActor])
      calculator ! op
    case result: MathResult => result match {
      case MultiplicationResult(n1, n2, r) =>
        printf("Mul result: %d * %d = %d\n", n1, n2, r)
        context.stop(sender())
      case DivisionResult(n1, n2, r) =>
        printf("Div result: %.0f / %d = %.2f\n", n1, n2, r)
        context.stop(sender())
    }
  }
} 

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

import akka.actor.ActorLogging
import akka.actor.Actor

//#metrics-listener
import akka.cluster.Cluster
import akka.cluster.ClusterEvent.ClusterMetricsChanged
import akka.cluster.ClusterEvent.CurrentClusterState
import akka.cluster.NodeMetrics
import akka.cluster.StandardMetrics.HeapMemory
import akka.cluster.StandardMetrics.Cpu

class MetricsListener extends Actor with ActorLogging {
  val selfAddress = Cluster(context.system).selfAddress

  // subscribe to ClusterMetricsChanged
  // re-subscribe when restart
  override def preStart(): Unit =
    Cluster(context.system).subscribe(self, classOf[ClusterMetricsChanged])
  override def postStop(): Unit =
    Cluster(context.system).unsubscribe(self)

  def receive = {
    case ClusterMetricsChanged(clusterMetrics) =>
      clusterMetrics.filter(_.address == selfAddress) foreach { nodeMetrics =>
        logHeap(nodeMetrics)
        logCpu(nodeMetrics)
      }
    case state: CurrentClusterState => // ignore
  }

  def logHeap(nodeMetrics: NodeMetrics): Unit = nodeMetrics match {
    case HeapMemory(address, timestamp, used, committed, max) =>
      log.info("Used heap: {} MB", used.doubleValue / 1024 / 1024)
    case _ => // no heap info
  }

  def logCpu(nodeMetrics: NodeMetrics): Unit = nodeMetrics match {
    case Cpu(address, timestamp, Some(systemLoadAverage), cpuCombined, processors) =>
      log.info("Load: {} ({} processors)", systemLoadAverage, processors)
    case _ => // no cpu info
  }
}

//#metrics-listener 

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

import akka.cluster.Cluster
import akka.cluster.ClusterEvent._
import akka.actor.ActorLogging
import akka.actor.Actor

class SimpleClusterListener extends Actor with ActorLogging {

  val cluster = Cluster(context.system)

  // subscribe to cluster changes, re-subscribe when restart 
  override def preStart(): Unit = {
    //#subscribe
    cluster.subscribe(self, initialStateMode = InitialStateAsEvents,
      classOf[MemberEvent], classOf[UnreachableMember])
    //#subscribe
  }
  override def postStop(): Unit = cluster.unsubscribe(self)

  def receive = {
    case MemberUp(member) =>
      log.info("Member is Up: {}", member.address)
    case UnreachableMember(member) =>
      log.info("Member detected as unreachable: {}", member)
    case MemberRemoved(member, previousStatus) =>
      log.info("Member is Removed: {} after {}",
        member.address, previousStatus)
    case _: MemberEvent => // ignore
  }
} 

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

import akka.cluster.Cluster
import akka.cluster.ClusterEvent._
import akka.actor.ActorLogging
import akka.actor.Actor

class SimpleClusterListener2 extends Actor with ActorLogging {

  val cluster = Cluster(context.system)

  // subscribe to cluster changes, re-subscribe when restart 
  override def preStart(): Unit = {
    //#subscribe
    cluster.subscribe(self, classOf[MemberEvent], classOf[UnreachableMember])
    //#subscribe
  }
  override def postStop(): Unit = cluster.unsubscribe(self)

  def receive = {
    case state: CurrentClusterState =>
      log.info("Current members: {}", state.members.mkString(", "))
    case MemberUp(member) =>
      log.info("Member is Up: {}", member.address)
    case UnreachableMember(member) =>
      log.info("Member detected as unreachable: {}", member)
    case MemberRemoved(member, previousStatus) =>
      log.info("Member is Removed: {} after {}",
        member.address, previousStatus)
    case _: MemberEvent => // ignore
  }
} 

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

import language.postfixOps
import scala.concurrent.duration._
import akka.actor.Actor
import akka.actor.ActorRef
import akka.actor.ActorSystem
import akka.actor.Props
import akka.actor.RootActorPath
import akka.cluster.Cluster
import akka.cluster.ClusterEvent.CurrentClusterState
import akka.cluster.ClusterEvent.MemberUp
import akka.cluster.Member
import akka.cluster.MemberStatus
import com.typesafe.config.ConfigFactory

//#backend
class TransformationBackend extends Actor {

  val cluster = Cluster(context.system)

  // subscribe to cluster changes, MemberUp
  // re-subscribe when restart
  override def preStart(): Unit = cluster.subscribe(self, classOf[MemberUp])
  override def postStop(): Unit = cluster.unsubscribe(self)

  def receive = {
    case TransformationJob(text) => sender() ! TransformationResult(text.toUpperCase)
    case state: CurrentClusterState =>
      state.members.filter(_.status == MemberStatus.Up) foreach register
    case MemberUp(m) => register(m)
  }

  def register(member: Member): Unit =
    if (member.hasRole("frontend"))
      context.actorSelection(RootActorPath(member.address) / "user" / "frontend") !
        BackendRegistration
}
//#backend

object TransformationBackend {
  def main(args: Array[String]): Unit = {
    // Override the configuration of the port when specified as program argument
    val port = if (args.isEmpty) "0" else args(0)
    val config = ConfigFactory.parseString(s"akka.remote.netty.tcp.port=$port").
      withFallback(ConfigFactory.parseString("akka.cluster.roles = [backend]")).
      withFallback(ConfigFactory.load())

    val system = ActorSystem("ClusterSystem", config)
    system.actorOf(Props[TransformationBackend], name = "backend")
  }
}