85 lines
2.8 KiB
Scala
85 lines
2.8 KiB
Scala
|
/**
|
||
|
|
* 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._
|
||
|
|
|
||
|
|
//#query-protocol
|
||
|
|
object DeviceGroup {
|
||
|
|
//#query-protocol
|
||
|
|
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
|
||
|
|
// ... earlier message types not shown
|
||
|
|
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
|