У меня есть отношения родитель -> дочерний субъект для загрузки файлов в Dropbox. Отношение состоит из актера-супервизора и актера загрузки. Субъект супервизора определяет стратегию супервизора для субъекта загрузки. Поэтому, если загрузка в Dropbox не удалась, актор должен быть перезапущен до тех пор, пока не будет достигнуто максимальное количество повторных попыток. В моем приложении меня интересует статус загрузки. Поэтому я использую шаблон «спросить», чтобы получить будущее в случае успеха или неудачи. Ниже вы можете найти текущую реализацию моих актеров.
/**
* An upload message.
*
* @param byte The byte array representing the content of a file.
* @param path The path under which the file should be stored.
*/
case class UploadMsg(byte: Array[Byte], path: String)
/**
* The upload supervisor.
*/
class UploadSupervisor extends Actor {
/**
* Stores the sender to the executing actor.
*/
val senders: ParHashMap[String, ActorRef] = ParHashMap()
/**
* Defines the supervisor strategy
*/
override val supervisorStrategy = OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1 minute) {
case _: DbxException => Restart
case e: Exception => Stop
}
/**
* Handles the received messages.
*/
override def receive: Actor.Receive = {
case msg: UploadMsg =>
implicit val timeout = Timeout(60.seconds)
val actor = context.actorOf(PropsContext.get(classOf[UploadActor]))
senders += actor.path.toString -> sender
context.watch(actor)
ask(actor, msg).mapTo[Unit] pipeTo sender
case Terminated(a) =>
context.unwatch(a)
senders.get(a.path.toString).map { sender =>
sender ! akka.actor.Status.Failure(new Exception("Actor terminated"))
senders - a.path.toString
}
}
}
/**
* An aktor which uploads a file to Dropbox.
*/
class UploadActor @Inject() (client: DropboxClient) extends Actor {
/**
* Sends the message again after restart.
*
* @param reason The reason why an restart occurred.
* @param message The message which causes the restart.
*/
override def preRestart(reason: Throwable, message: Option[Any]): Unit = {
super.preRestart(reason, message)
message foreach { self forward }
}
/**
* Handles the received messages.
*/
override def receive: Receive = {
case UploadMsg(byte, path) =>
val encrypted = encryptor.encrypt(byte)
val is = new ByteArrayInputStream(encrypted)
try {
client.storeFile("/" + path, DbxWriteMode.force(), encrypted.length, is)
sender ! (())
} finally {
is.close()
}
}
}
У меня вопрос: есть ли лучшее решение, чтобы сообщить моему приложению, что субъект загрузки завершился неудачно после указанного числа или повторных попыток. Особенно неудобно выглядит карта для хранения отправителей для актеров.
