scala - Akka Actor - メッセージが届くまでしばらく待ち、それ以外の場合はメッセージを送信します

Question

メッセージを受信するActorまで X 秒待機し、メッセージを受信した場合は通常どおり処理し、それ以外の場合は別のメッセージActor(コンストラクターで事前に決定) を送信することはできますか?

Answer 1

可能です。Akka Actor の「ask」と「Await」を TimeoutException で確認してください。ただし、アクター内でブロックすることは非常に悪い考えであることに注意してください。その間、アクターは他のメッセージを処理できないからです。さらに、1 つの Akka 処理スレッドをブロックします。

より良いアプローチは、メッセージを送信し (ファイア アンド フォーゲット)、 Akka スケジューラを使用してタイムアウト イベントをスケジュールすることです。応答が到着したら、そのイベントをキャンセルするか、何らかのフラグを設定して、応答が実際に時間どおりに到着した場合にトリガーされないようにします。

Answer 2

はい、メッセージを待ちたい場合は、単純にreceiveTimeoutを設定します: http://doc.akka.io/docs/akka/current/scala/actors.html#receive-timeout

(ドキュメントはここで少し誤解を招きます。すべてのメッセージの後に receiveTimeout を設定することもできます)

Answer 3

やり過ぎかもしれませんが、Finite State Machine (FSM)トレイトをチェックしてみてください。

import akka._
import actor._
import util._
import duration._
import Impatient._

object Impatient {
  sealed trait State
  case object WaitingForMessage extends State
  case object MessageReceived extends State
  case object TimeoutExpired extends State

  sealed trait Data
  case object Unitialized extends Data

  // In
  case object Message
}

class Impatient(receiver: ActorRef) extends Actor with FSM[State, Data] {
  startWith(WaitingForMessage, Unitialized)

  when(WaitingForMessage, stateTimeout = 3 seconds) {
    case Event(StateTimeout, data) => goto(TimeoutExpired) using data // data is usually modified here
    case Event(Message, data) => goto(MessageReceived) using data // data is usually modified here
  }

  onTransition {
    case WaitingForMessage -> MessageReceived => stateData match {
      case data => log.info("Received message: " + data)
    }
    case WaitingForMessage -> TimeoutExpired => receiver ! TimeoutExpired
  }

  when(MessageReceived) {
    case _ => stay
  }

  when(TimeoutExpired) {
    case _ => stay
  }

  initialize
}

これが実際の動作です：

object Main extends App {
  import akka._
  import actor._
  import Impatient._

  val system = ActorSystem("System")

  val receiver = system.actorOf(Props(new Actor with ActorLogging {
    def receive = {
      case TimeoutExpired => log.warning("Timeout expired")
    }
  }))

  val impatient = system.actorOf(Props(new Impatient(receiver)), name = "Impatient")
  impatient ! Message

  val impatient2 = system.actorOf(Props(new Impatient(receiver)), name = "Impatient2")
  Thread.sleep(4000)
  impatient2 ! Message

  system.shutdown()
}