Move content from akka-docs-new into akka-docs
Now the paradox documentation is no longer functional until we update akka-docs to generate from paradox instead of sphinx
This commit is contained in:
Arnout Engelen
parent
f064d1321a
commit
5507147073
96 changed files with 0 additions and 35 deletions
46
akka-docs/src/test/scala/tutorial_4/Device.scala
Normal file
46
akka-docs/src/test/scala/tutorial_4/Device.scala
Normal file
|
|
@ -0,0 +1,46 @@
|
|||
/**
|
||||
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
|
||||
*/
|
||||
package tutorial_4
|
||||
|
||||
import akka.actor.{ Actor, ActorLogging, Props }
|
||||
import tutorial_4.Device.{ ReadTemperature, RecordTemperature, RespondTemperature, TemperatureRecorded }
|
||||
import tutorial_4.DeviceManager.{ DeviceRegistered, RequestTrackDevice }
|
||||
|
||||
object Device {
|
||||
|
||||
def props(groupId: String, deviceId: String): Props = Props(new Device(groupId, deviceId))
|
||||
|
||||
final case class RecordTemperature(requestId: Long, value: Double)
|
||||
final case class TemperatureRecorded(requestId: Long)
|
||||
|
||||
final case class ReadTemperature(requestId: Long)
|
||||
final case class RespondTemperature(requestId: Long, value: Option[Double])
|
||||
}
|
||||
|
||||
class Device(groupId: String, deviceId: String) extends Actor with ActorLogging {
|
||||
var lastTemperatureReading: Option[Double] = None
|
||||
|
||||
override def preStart(): Unit = log.info("Device actor {}-{} started", groupId, deviceId)
|
||||
|
||||
override def postStop(): Unit = log.info("Device actor {}-{} stopped", groupId, deviceId)
|
||||
|
||||
override def receive: Receive = {
|
||||
case RequestTrackDevice(`groupId`, `deviceId`) =>
|
||||
sender() ! DeviceRegistered
|
||||
|
||||
case RequestTrackDevice(groupId, deviceId) =>
|
||||
log.warning(
|
||||
"Ignoring TrackDevice request for {}-{}.This actor is responsible for {}-{}.",
|
||||
groupId, deviceId, this.groupId, this.deviceId
|
||||
)
|
||||
|
||||
case RecordTemperature(id, value) =>
|
||||
log.info("Recorded temperature reading {} with {}", value, id)
|
||||
lastTemperatureReading = Some(value)
|
||||
sender() ! TemperatureRecorded(id)
|
||||
|
||||
case ReadTemperature(id) =>
|
||||
sender() ! RespondTemperature(id, lastTemperatureReading)
|
||||
}
|
||||
}
|
||||
83
akka-docs/src/test/scala/tutorial_4/DeviceGroup.scala
Normal file
83
akka-docs/src/test/scala/tutorial_4/DeviceGroup.scala
Normal file
|
|
@ -0,0 +1,83 @@
|
|||
/**
|
||||
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
|
||||
*/
|
||||
package tutorial_4
|
||||
|
||||
import akka.actor.{ Actor, ActorLogging, ActorRef, Props, Terminated }
|
||||
import tutorial_4.DeviceGroup._
|
||||
import tutorial_4.DeviceManager.RequestTrackDevice
|
||||
|
||||
import scala.concurrent.duration._
|
||||
|
||||
object DeviceGroup {
|
||||
def props(groupId: String): Props = Props(new DeviceGroup(groupId))
|
||||
|
||||
final case class RequestDeviceList(requestId: Long)
|
||||
final case class ReplyDeviceList(requestId: Long, ids: Set[String])
|
||||
|
||||
//#query-protocol
|
||||
final case class RequestAllTemperatures(requestId: Long)
|
||||
final case class RespondAllTemperatures(requestId: Long, temperatures: Map[String, TemperatureReading])
|
||||
|
||||
sealed trait TemperatureReading
|
||||
final case class Temperature(value: Double) extends TemperatureReading
|
||||
case object TemperatureNotAvailable extends TemperatureReading
|
||||
case object DeviceNotAvailable extends TemperatureReading
|
||||
case object DeviceTimedOut extends TemperatureReading
|
||||
//#query-protocol
|
||||
}
|
||||
|
||||
//#query-added
|
||||
class DeviceGroup(groupId: String) extends Actor with ActorLogging {
|
||||
var deviceIdToActor = Map.empty[String, ActorRef]
|
||||
var actorToDeviceId = Map.empty[ActorRef, String]
|
||||
var nextCollectionId = 0L
|
||||
|
||||
override def preStart(): Unit = log.info("DeviceGroup {} started", groupId)
|
||||
|
||||
override def postStop(): Unit = log.info("DeviceGroup {} stopped", groupId)
|
||||
|
||||
override def receive: Receive = {
|
||||
//#query-added
|
||||
case trackMsg @ RequestTrackDevice(`groupId`, _) =>
|
||||
deviceIdToActor.get(trackMsg.deviceId) match {
|
||||
case Some(ref) =>
|
||||
ref forward trackMsg
|
||||
case None =>
|
||||
log.info("Creating device actor for {}", trackMsg.deviceId)
|
||||
val deviceActor = context.actorOf(Device.props(groupId, trackMsg.deviceId), "device-" + trackMsg.deviceId)
|
||||
context.watch(deviceActor)
|
||||
deviceActor forward trackMsg
|
||||
deviceIdToActor += trackMsg.deviceId -> deviceActor
|
||||
actorToDeviceId += deviceActor -> trackMsg.deviceId
|
||||
}
|
||||
|
||||
case RequestTrackDevice(groupId, deviceId) =>
|
||||
log.warning(
|
||||
"Ignoring TrackDevice request for {}. This actor is responsible for {}.",
|
||||
groupId, this.groupId
|
||||
)
|
||||
|
||||
case RequestDeviceList(requestId) =>
|
||||
sender() ! ReplyDeviceList(requestId, deviceIdToActor.keySet)
|
||||
|
||||
case Terminated(deviceActor) =>
|
||||
val deviceId = actorToDeviceId(deviceActor)
|
||||
log.info("Device actor for {} has been terminated", deviceId)
|
||||
actorToDeviceId -= deviceActor
|
||||
deviceIdToActor -= deviceId
|
||||
|
||||
//#query-added
|
||||
// ... other cases omitted
|
||||
|
||||
case RequestAllTemperatures(requestId) =>
|
||||
context.actorOf(DeviceGroupQuery.props(
|
||||
actorToDeviceId = actorToDeviceId,
|
||||
requestId = requestId,
|
||||
requester = sender(),
|
||||
3.seconds
|
||||
))
|
||||
}
|
||||
|
||||
}
|
||||
//#query-added
|
||||
104
akka-docs/src/test/scala/tutorial_4/DeviceGroupQuery.scala
Normal file
104
akka-docs/src/test/scala/tutorial_4/DeviceGroupQuery.scala
Normal file
|
|
@ -0,0 +1,104 @@
|
|||
/**
|
||||
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
|
||||
*/
|
||||
package tutorial_4
|
||||
|
||||
import akka.actor.{ Actor, ActorLogging, ActorRef, Props, Terminated }
|
||||
|
||||
import scala.concurrent.duration._
|
||||
|
||||
//#query-full
|
||||
//#query-outline
|
||||
object DeviceGroupQuery {
|
||||
case object CollectionTimeout
|
||||
|
||||
def props(
|
||||
actorToDeviceId: Map[ActorRef, String],
|
||||
requestId: Long,
|
||||
requester: ActorRef,
|
||||
timeout: FiniteDuration
|
||||
): Props = {
|
||||
Props(new DeviceGroupQuery(actorToDeviceId, requestId, requester, timeout))
|
||||
}
|
||||
}
|
||||
|
||||
class DeviceGroupQuery(
|
||||
actorToDeviceId: Map[ActorRef, String],
|
||||
requestId: Long,
|
||||
requester: ActorRef,
|
||||
timeout: FiniteDuration
|
||||
) extends Actor with ActorLogging {
|
||||
import DeviceGroupQuery._
|
||||
import context.dispatcher
|
||||
val queryTimeoutTimer = context.system.scheduler.scheduleOnce(timeout, self, CollectionTimeout)
|
||||
|
||||
override def preStart(): Unit = {
|
||||
actorToDeviceId.keysIterator.foreach { deviceActor =>
|
||||
context.watch(deviceActor)
|
||||
deviceActor ! Device.ReadTemperature(0)
|
||||
}
|
||||
}
|
||||
|
||||
override def postStop(): Unit = {
|
||||
queryTimeoutTimer.cancel()
|
||||
}
|
||||
|
||||
//#query-outline
|
||||
//#query-state
|
||||
override def receive: Receive =
|
||||
waitingForReplies(
|
||||
Map.empty,
|
||||
actorToDeviceId.keySet
|
||||
)
|
||||
|
||||
def waitingForReplies(
|
||||
repliesSoFar: Map[String, DeviceGroup.TemperatureReading],
|
||||
stillWaiting: Set[ActorRef]
|
||||
): Receive = {
|
||||
case Device.RespondTemperature(0, valueOption) =>
|
||||
val deviceActor = sender()
|
||||
val reading = valueOption match {
|
||||
case Some(value) => DeviceGroup.Temperature(value)
|
||||
case None => DeviceGroup.TemperatureNotAvailable
|
||||
}
|
||||
receivedResponse(deviceActor, reading, stillWaiting, repliesSoFar)
|
||||
|
||||
case Terminated(deviceActor) =>
|
||||
receivedResponse(deviceActor, DeviceGroup.DeviceNotAvailable, stillWaiting, repliesSoFar)
|
||||
|
||||
case CollectionTimeout =>
|
||||
val timedOutReplies =
|
||||
stillWaiting.map { deviceActor =>
|
||||
val deviceId = actorToDeviceId(deviceActor)
|
||||
deviceId -> DeviceGroup.DeviceTimedOut
|
||||
}
|
||||
requester ! DeviceGroup.RespondAllTemperatures(requestId, repliesSoFar ++ timedOutReplies)
|
||||
context.stop(self)
|
||||
}
|
||||
//#query-state
|
||||
|
||||
//#query-collect-reply
|
||||
def receivedResponse(
|
||||
deviceActor: ActorRef,
|
||||
reading: DeviceGroup.TemperatureReading,
|
||||
stillWaiting: Set[ActorRef],
|
||||
repliesSoFar: Map[String, DeviceGroup.TemperatureReading]
|
||||
): Unit = {
|
||||
context.unwatch(deviceActor)
|
||||
val deviceId = actorToDeviceId(deviceActor)
|
||||
val newStillWaiting = stillWaiting - deviceActor
|
||||
|
||||
val newRepliesSoFar = repliesSoFar + (deviceId -> reading)
|
||||
if (newStillWaiting.isEmpty) {
|
||||
requester ! DeviceGroup.RespondAllTemperatures(requestId, newRepliesSoFar)
|
||||
context.stop(self)
|
||||
} else {
|
||||
context.become(waitingForReplies(newRepliesSoFar, newStillWaiting))
|
||||
}
|
||||
}
|
||||
//#query-collect-reply
|
||||
|
||||
//#query-outline
|
||||
}
|
||||
//#query-outline
|
||||
//#query-full
|
||||
167
akka-docs/src/test/scala/tutorial_4/DeviceGroupQuerySpec.scala
Normal file
167
akka-docs/src/test/scala/tutorial_4/DeviceGroupQuerySpec.scala
Normal file
|
|
@ -0,0 +1,167 @@
|
|||
/**
|
||||
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
|
||||
*/
|
||||
package tutorial_4
|
||||
|
||||
import akka.actor.PoisonPill
|
||||
import akka.testkit.{ AkkaSpec, TestProbe }
|
||||
|
||||
import scala.concurrent.duration._
|
||||
|
||||
class DeviceGroupQuerySpec extends AkkaSpec {
|
||||
|
||||
"DeviceGroupQuery" must {
|
||||
|
||||
//#query-test-normal
|
||||
"return temperature value for working devices" in {
|
||||
val requester = TestProbe()
|
||||
|
||||
val device1 = TestProbe()
|
||||
val device2 = TestProbe()
|
||||
|
||||
val queryActor = system.actorOf(DeviceGroupQuery.props(
|
||||
actorToDeviceId = Map(device1.ref -> "device1", device2.ref -> "device2"),
|
||||
requestId = 1,
|
||||
requester = requester.ref,
|
||||
timeout = 3.seconds
|
||||
))
|
||||
|
||||
device1.expectMsg(Device.ReadTemperature(requestId = 0))
|
||||
device2.expectMsg(Device.ReadTemperature(requestId = 0))
|
||||
|
||||
queryActor.tell(Device.RespondTemperature(requestId = 0, Some(1.0)), device1.ref)
|
||||
queryActor.tell(Device.RespondTemperature(requestId = 0, Some(2.0)), device2.ref)
|
||||
|
||||
requester.expectMsg(DeviceGroup.RespondAllTemperatures(
|
||||
requestId = 1,
|
||||
temperatures = Map(
|
||||
"device1" -> DeviceGroup.Temperature(1.0),
|
||||
"device2" -> DeviceGroup.Temperature(2.0)
|
||||
)
|
||||
))
|
||||
}
|
||||
//#query-test-normal
|
||||
|
||||
//#query-test-no-reading
|
||||
"return TemperatureNotAvailable for devices with no readings" in {
|
||||
val requester = TestProbe()
|
||||
|
||||
val device1 = TestProbe()
|
||||
val device2 = TestProbe()
|
||||
|
||||
val queryActor = system.actorOf(DeviceGroupQuery.props(
|
||||
actorToDeviceId = Map(device1.ref -> "device1", device2.ref -> "device2"),
|
||||
requestId = 1,
|
||||
requester = requester.ref,
|
||||
timeout = 3.seconds
|
||||
))
|
||||
|
||||
device1.expectMsg(Device.ReadTemperature(requestId = 0))
|
||||
device2.expectMsg(Device.ReadTemperature(requestId = 0))
|
||||
|
||||
queryActor.tell(Device.RespondTemperature(requestId = 0, None), device1.ref)
|
||||
queryActor.tell(Device.RespondTemperature(requestId = 0, Some(2.0)), device2.ref)
|
||||
|
||||
requester.expectMsg(DeviceGroup.RespondAllTemperatures(
|
||||
requestId = 1,
|
||||
temperatures = Map(
|
||||
"device1" -> DeviceGroup.TemperatureNotAvailable,
|
||||
"device2" -> DeviceGroup.Temperature(2.0)
|
||||
)
|
||||
))
|
||||
}
|
||||
//#query-test-no-reading
|
||||
|
||||
//#query-test-stopped
|
||||
"return DeviceNotAvailable if device stops before answering" in {
|
||||
val requester = TestProbe()
|
||||
|
||||
val device1 = TestProbe()
|
||||
val device2 = TestProbe()
|
||||
|
||||
val queryActor = system.actorOf(DeviceGroupQuery.props(
|
||||
actorToDeviceId = Map(device1.ref -> "device1", device2.ref -> "device2"),
|
||||
requestId = 1,
|
||||
requester = requester.ref,
|
||||
timeout = 3.seconds
|
||||
))
|
||||
|
||||
device1.expectMsg(Device.ReadTemperature(requestId = 0))
|
||||
device2.expectMsg(Device.ReadTemperature(requestId = 0))
|
||||
|
||||
queryActor.tell(Device.RespondTemperature(requestId = 0, Some(1.0)), device1.ref)
|
||||
device2.ref ! PoisonPill
|
||||
|
||||
requester.expectMsg(DeviceGroup.RespondAllTemperatures(
|
||||
requestId = 1,
|
||||
temperatures = Map(
|
||||
"device1" -> DeviceGroup.Temperature(1.0),
|
||||
"device2" -> DeviceGroup.DeviceNotAvailable
|
||||
)
|
||||
))
|
||||
}
|
||||
//#query-test-stopped
|
||||
|
||||
//#query-test-stopped-later
|
||||
"return temperature reading even if device stops after answering" in {
|
||||
val requester = TestProbe()
|
||||
|
||||
val device1 = TestProbe()
|
||||
val device2 = TestProbe()
|
||||
|
||||
val queryActor = system.actorOf(DeviceGroupQuery.props(
|
||||
actorToDeviceId = Map(device1.ref -> "device1", device2.ref -> "device2"),
|
||||
requestId = 1,
|
||||
requester = requester.ref,
|
||||
timeout = 3.seconds
|
||||
))
|
||||
|
||||
device1.expectMsg(Device.ReadTemperature(requestId = 0))
|
||||
device2.expectMsg(Device.ReadTemperature(requestId = 0))
|
||||
|
||||
queryActor.tell(Device.RespondTemperature(requestId = 0, Some(1.0)), device1.ref)
|
||||
queryActor.tell(Device.RespondTemperature(requestId = 0, Some(2.0)), device2.ref)
|
||||
device2.ref ! PoisonPill
|
||||
|
||||
requester.expectMsg(DeviceGroup.RespondAllTemperatures(
|
||||
requestId = 1,
|
||||
temperatures = Map(
|
||||
"device1" -> DeviceGroup.Temperature(1.0),
|
||||
"device2" -> DeviceGroup.Temperature(2.0)
|
||||
)
|
||||
))
|
||||
}
|
||||
//#query-test-stopped-later
|
||||
|
||||
//#query-test-timeout
|
||||
"return DeviceTimedOut if device does not answer in time" in {
|
||||
val requester = TestProbe()
|
||||
|
||||
val device1 = TestProbe()
|
||||
val device2 = TestProbe()
|
||||
|
||||
val queryActor = system.actorOf(DeviceGroupQuery.props(
|
||||
actorToDeviceId = Map(device1.ref -> "device1", device2.ref -> "device2"),
|
||||
requestId = 1,
|
||||
requester = requester.ref,
|
||||
timeout = 1.second
|
||||
))
|
||||
|
||||
device1.expectMsg(Device.ReadTemperature(requestId = 0))
|
||||
device2.expectMsg(Device.ReadTemperature(requestId = 0))
|
||||
|
||||
queryActor.tell(Device.RespondTemperature(requestId = 0, Some(1.0)), device1.ref)
|
||||
|
||||
requester.expectMsg(DeviceGroup.RespondAllTemperatures(
|
||||
requestId = 1,
|
||||
temperatures = Map(
|
||||
"device1" -> DeviceGroup.Temperature(1.0),
|
||||
"device2" -> DeviceGroup.DeviceTimedOut
|
||||
)
|
||||
))
|
||||
}
|
||||
//#query-test-timeout
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
135
akka-docs/src/test/scala/tutorial_4/DeviceGroupSpec.scala
Normal file
135
akka-docs/src/test/scala/tutorial_4/DeviceGroupSpec.scala
Normal file
|
|
@ -0,0 +1,135 @@
|
|||
/**
|
||||
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
|
||||
*/
|
||||
|
||||
package tutorial_4
|
||||
|
||||
import akka.actor.PoisonPill
|
||||
import akka.testkit.{ AkkaSpec, TestProbe }
|
||||
|
||||
import scala.concurrent.duration._
|
||||
|
||||
class DeviceGroupSpec extends AkkaSpec {
|
||||
|
||||
"DeviceGroup actor" must {
|
||||
|
||||
"be able to register a device actor" in {
|
||||
val probe = TestProbe()
|
||||
val groupActor = system.actorOf(DeviceGroup.props("group"))
|
||||
|
||||
groupActor.tell(DeviceManager.RequestTrackDevice("group", "device1"), probe.ref)
|
||||
probe.expectMsg(DeviceManager.DeviceRegistered)
|
||||
val deviceActor1 = probe.lastSender
|
||||
|
||||
groupActor.tell(DeviceManager.RequestTrackDevice("group", "device2"), probe.ref)
|
||||
probe.expectMsg(DeviceManager.DeviceRegistered)
|
||||
val deviceActor2 = probe.lastSender
|
||||
deviceActor1 should !==(deviceActor2)
|
||||
|
||||
// Check that the device actors are working
|
||||
deviceActor1.tell(Device.RecordTemperature(requestId = 0, 1.0), probe.ref)
|
||||
probe.expectMsg(Device.TemperatureRecorded(requestId = 0))
|
||||
deviceActor2.tell(Device.RecordTemperature(requestId = 1, 2.0), probe.ref)
|
||||
probe.expectMsg(Device.TemperatureRecorded(requestId = 1))
|
||||
}
|
||||
|
||||
"ignore requests for wrong groupId" in {
|
||||
val probe = TestProbe()
|
||||
val groupActor = system.actorOf(DeviceGroup.props("group"))
|
||||
|
||||
groupActor.tell(DeviceManager.RequestTrackDevice("wrongGroup", "device1"), probe.ref)
|
||||
probe.expectNoMsg(500.milliseconds)
|
||||
}
|
||||
|
||||
"return same actor for same deviceId" in {
|
||||
val probe = TestProbe()
|
||||
val groupActor = system.actorOf(DeviceGroup.props("group"))
|
||||
|
||||
groupActor.tell(DeviceManager.RequestTrackDevice("group", "device1"), probe.ref)
|
||||
probe.expectMsg(DeviceManager.DeviceRegistered)
|
||||
val deviceActor1 = probe.lastSender
|
||||
|
||||
groupActor.tell(DeviceManager.RequestTrackDevice("group", "device1"), probe.ref)
|
||||
probe.expectMsg(DeviceManager.DeviceRegistered)
|
||||
val deviceActor2 = probe.lastSender
|
||||
|
||||
deviceActor1 should ===(deviceActor2)
|
||||
}
|
||||
|
||||
"be able to list active devices" in {
|
||||
val probe = TestProbe()
|
||||
val groupActor = system.actorOf(DeviceGroup.props("group"))
|
||||
|
||||
groupActor.tell(DeviceManager.RequestTrackDevice("group", "device1"), probe.ref)
|
||||
probe.expectMsg(DeviceManager.DeviceRegistered)
|
||||
|
||||
groupActor.tell(DeviceManager.RequestTrackDevice("group", "device2"), probe.ref)
|
||||
probe.expectMsg(DeviceManager.DeviceRegistered)
|
||||
|
||||
groupActor.tell(DeviceGroup.RequestDeviceList(requestId = 0), probe.ref)
|
||||
probe.expectMsg(DeviceGroup.ReplyDeviceList(requestId = 0, Set("device1", "device2")))
|
||||
}
|
||||
|
||||
"be able to list active devices after one shuts down" in {
|
||||
val probe = TestProbe()
|
||||
val groupActor = system.actorOf(DeviceGroup.props("group"))
|
||||
|
||||
groupActor.tell(DeviceManager.RequestTrackDevice("group", "device1"), probe.ref)
|
||||
probe.expectMsg(DeviceManager.DeviceRegistered)
|
||||
val toShutDown = probe.lastSender
|
||||
|
||||
groupActor.tell(DeviceManager.RequestTrackDevice("group", "device2"), probe.ref)
|
||||
probe.expectMsg(DeviceManager.DeviceRegistered)
|
||||
|
||||
groupActor.tell(DeviceGroup.RequestDeviceList(requestId = 0), probe.ref)
|
||||
probe.expectMsg(DeviceGroup.ReplyDeviceList(requestId = 0, Set("device1", "device2")))
|
||||
|
||||
probe.watch(toShutDown)
|
||||
toShutDown ! PoisonPill
|
||||
probe.expectTerminated(toShutDown)
|
||||
|
||||
groupActor.tell(DeviceGroup.RequestDeviceList(requestId = 1), probe.ref)
|
||||
probe.expectMsg(DeviceGroup.ReplyDeviceList(requestId = 1, Set("device2")))
|
||||
}
|
||||
|
||||
//#group-query-integration-test
|
||||
"be able to collect temperatures from all active devices" in {
|
||||
val probe = TestProbe()
|
||||
val groupActor = system.actorOf(DeviceGroup.props("group"))
|
||||
|
||||
groupActor.tell(DeviceManager.RequestTrackDevice("group", "device1"), probe.ref)
|
||||
probe.expectMsg(DeviceManager.DeviceRegistered)
|
||||
val deviceActor1 = probe.lastSender
|
||||
|
||||
groupActor.tell(DeviceManager.RequestTrackDevice("group", "device2"), probe.ref)
|
||||
probe.expectMsg(DeviceManager.DeviceRegistered)
|
||||
val deviceActor2 = probe.lastSender
|
||||
|
||||
groupActor.tell(DeviceManager.RequestTrackDevice("group", "device3"), probe.ref)
|
||||
probe.expectMsg(DeviceManager.DeviceRegistered)
|
||||
val deviceActor3 = probe.lastSender
|
||||
|
||||
// Check that the device actors are working
|
||||
deviceActor1.tell(Device.RecordTemperature(requestId = 0, 1.0), probe.ref)
|
||||
probe.expectMsg(Device.TemperatureRecorded(requestId = 0))
|
||||
deviceActor2.tell(Device.RecordTemperature(requestId = 1, 2.0), probe.ref)
|
||||
probe.expectMsg(Device.TemperatureRecorded(requestId = 1))
|
||||
// No temperature for device3
|
||||
|
||||
groupActor.tell(DeviceGroup.RequestAllTemperatures(requestId = 0), probe.ref)
|
||||
probe.expectMsg(
|
||||
DeviceGroup.RespondAllTemperatures(
|
||||
requestId = 0,
|
||||
temperatures = Map(
|
||||
"device1" -> DeviceGroup.Temperature(1.0),
|
||||
"device2" -> DeviceGroup.Temperature(2.0),
|
||||
"device3" -> DeviceGroup.TemperatureNotAvailable
|
||||
)
|
||||
)
|
||||
)
|
||||
}
|
||||
//#group-query-integration-test
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
47
akka-docs/src/test/scala/tutorial_4/DeviceManager.scala
Normal file
47
akka-docs/src/test/scala/tutorial_4/DeviceManager.scala
Normal file
|
|
@ -0,0 +1,47 @@
|
|||
/**
|
||||
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
|
||||
*/
|
||||
|
||||
package tutorial_4
|
||||
|
||||
import akka.actor.{ Actor, ActorLogging, ActorRef, Props, Terminated }
|
||||
import tutorial_4.DeviceManager.RequestTrackDevice
|
||||
|
||||
object DeviceManager {
|
||||
def props(): Props = Props(new DeviceManager)
|
||||
|
||||
final case class RequestTrackDevice(groupId: String, deviceId: String)
|
||||
case object DeviceRegistered
|
||||
}
|
||||
|
||||
class DeviceManager extends Actor with ActorLogging {
|
||||
var groupIdToActor = Map.empty[String, ActorRef]
|
||||
var actorToGroupId = Map.empty[ActorRef, String]
|
||||
|
||||
override def preStart(): Unit = log.info("DeviceManager started")
|
||||
|
||||
override def postStop(): Unit = log.info("DeviceManager stopped")
|
||||
|
||||
override def receive = {
|
||||
case trackMsg @ RequestTrackDevice(groupId, _) =>
|
||||
groupIdToActor.get(groupId) match {
|
||||
case Some(ref) =>
|
||||
ref forward trackMsg
|
||||
case None =>
|
||||
log.info("Creating device group actor for {}", groupId)
|
||||
val groupActor = context.actorOf(DeviceGroup.props(groupId), "group-" + groupId)
|
||||
context.watch(groupActor)
|
||||
groupActor forward trackMsg
|
||||
groupIdToActor += groupId -> groupActor
|
||||
actorToGroupId += groupActor -> groupId
|
||||
}
|
||||
|
||||
case Terminated(groupActor) =>
|
||||
val groupId = actorToGroupId(groupActor)
|
||||
log.info("Device group actor for {} has been terminated", groupId)
|
||||
actorToGroupId -= groupActor
|
||||
groupIdToActor -= groupId
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue