|
23 | 23 | * void Receive(TMessageId messageId, TBlob blob, TActorContext::TPtr ctx) override { |
24 | 24 | * if (messageId == MyMessage::MessageId) { |
25 | 25 | * auto message = DeserializeNear<MyMessage>(blob); |
26 | | - * // Process message |
27 | | - * ctx->Send(ctx->Sender(), ResponseMessage{}); |
| 26 | + * // Process message and reply to the sender |
| 27 | + * ctx->Send<ResponseMessage>(ctx->Sender(), message.value * 2); |
28 | 28 | * } |
29 | 29 | * } |
30 | 30 | * }; |
|
37 | 37 | * TFuture<void> CoReceive(TMessageId messageId, TBlob blob, TActorContext::TPtr ctx) override { |
38 | 38 | * if (messageId == MyMessage::MessageId) { |
39 | 39 | * auto message = DeserializeNear<MyMessage>(blob); |
| 40 | + * auto sender = ctx->Sender(); |
40 | 41 | * |
41 | | - * // Async operations |
| 42 | + * // Async sleep — mailbox is paused until this handler finishes |
42 | 43 | * co_await ctx->Sleep(std::chrono::seconds(1)); |
43 | 44 | * |
| 45 | + * // Request-reply: blocks this handler, not the event loop |
44 | 46 | * auto response = co_await ctx->Ask<ResponseMessage>( |
45 | 47 | * someActor, QueryMessage{} |
46 | 48 | * ); |
47 | 49 | * |
48 | | - * ctx->Send(ctx->Sender(), response); |
| 50 | + * ctx->Send<ResponseMessage>(sender, std::move(response)); |
49 | 51 | * } |
50 | 52 | * } |
51 | 53 | * }; |
|
88 | 90 | * AsyncBehaviorActor() { Become(this); } |
89 | 91 | * |
90 | 92 | * TFuture<void> Receive(ProcessMessage&& msg, TBlob blob, TActorContext::TPtr ctx) { |
91 | | - * // This method returns a future, so it will be handled asynchronously |
| 93 | + * // TFuture<void> return type triggers async dispatch automatically |
92 | 94 | * co_await ctx->Sleep(std::chrono::milliseconds(100)); |
93 | 95 | * |
94 | 96 | * auto result = co_await processAsync(msg.data); |
95 | | - * ctx->Send(ctx->Sender(), ResultMessage{result}); |
| 97 | + * ctx->Send<ResultMessage>(ctx->Sender(), std::move(result)); |
96 | 98 | * } |
97 | 99 | * |
98 | 100 | * void HandleUnknownMessage(TMessageId messageId, TBlob blob, TActorContext::TPtr ctx) { |
@@ -127,13 +129,17 @@ using TEvent = std::pair<unsigned, TTime>; |
127 | 129 | /** |
128 | 130 | * @brief Context object providing actor communication and scheduling capabilities |
129 | 131 | * |
130 | | - * TActorContext is passed to actors when they receive messages. It provides |
| 132 | + * TActorContext is passed to actors during message dispatch. It provides |
131 | 133 | * methods for sending messages, scheduling future messages, sleeping, and |
132 | 134 | * performing request-response patterns. |
133 | 135 | * |
134 | | - * The context contains information about the current message sender and |
135 | | - * the actor's own ID. It also manages asynchronous operations through |
136 | | - * the actor system. |
| 136 | + * The context is valid only for the duration of the `Receive` / `CoReceive` |
| 137 | + * call chain. Capture `ctx->Sender()` / `ctx->Self()` by value if you need |
| 138 | + * them after the first `co_await`. |
| 139 | + * |
| 140 | + * @note `Sleep`, `Ask`, and any coroutine-based methods require `co_await` |
| 141 | + * and are only meaningful inside `ICoroActor::CoReceive` or an async |
| 142 | + * `TBehavior::Receive` (one returning `TFuture<void>`). |
137 | 143 | */ |
138 | 144 | class TActorContext |
139 | 145 | { |
@@ -167,19 +173,32 @@ class TActorContext |
167 | 173 | void Forward(TActorId to, TMessageId messageId, TBlob blob); |
168 | 174 |
|
169 | 175 | /** |
170 | | - * @brief Send a typed message to another actor |
171 | | - * @tparam T Message type that has MessageId static member |
| 176 | + * @brief Send a typed message to another actor (non-blocking) |
| 177 | + * |
| 178 | + * Constructs a `T` in-place from `args` and enqueues it to the recipient. |
| 179 | + * Returns immediately; delivery is deferred to the next event-loop iteration. |
| 180 | + * |
| 181 | + * @tparam T Message type; must have a `static TMessageId MessageId` member |
172 | 182 | * @param to Recipient actor ID |
173 | | - * @param args passed to Message constructor |
| 183 | + * @param args Constructor arguments forwarded to `T` |
| 184 | + * |
| 185 | + * @code |
| 186 | + * ctx->Send<Pong>(ctx->Sender(), 42); // Pong{42} delivered to sender |
| 187 | + * @endcode |
174 | 188 | */ |
175 | 189 | template<typename T, typename... Args> |
176 | 190 | void Send(TActorId to, Args&&... args); |
177 | 191 |
|
178 | 192 | /** |
179 | | - * @brief Forward a typed message to another actor |
180 | | - * @tparam T Message type that has MessageId static member |
| 193 | + * @brief Forward a typed message to another actor, preserving the original sender |
| 194 | + * |
| 195 | + * Like `Send<T>`, but the recipient sees `ctx->Sender()` equal to the |
| 196 | + * sender of the *current* message rather than this actor's ID. Useful for |
| 197 | + * routing/proxy actors that should not appear in the reply chain. |
| 198 | + * |
| 199 | + * @tparam T Message type; must have a `static TMessageId MessageId` member |
181 | 200 | * @param to Recipient actor ID |
182 | | - * @param args passed to Message constructor |
| 201 | + * @param args Constructor arguments forwarded to `T` |
183 | 202 | */ |
184 | 203 | template<typename T, typename... Args> |
185 | 204 | void Forward(TActorId to, Args&&... args); |
@@ -214,29 +233,50 @@ class TActorContext |
214 | 233 | void Cancel(TEvent event); |
215 | 234 |
|
216 | 235 | /** |
217 | | - * @brief Sleep until a specific time |
218 | | - * @param until Time to sleep until |
219 | | - * @return Future that completes when the sleep time is reached |
| 236 | + * @brief Suspend the current handler until a specific time |
| 237 | + * |
| 238 | + * Must be `co_await`-ed inside `ICoroActor::CoReceive` or an async |
| 239 | + * `TBehavior::Receive`. The actor system continues processing other |
| 240 | + * actors while this handler is suspended. |
| 241 | + * |
| 242 | + * @param until Absolute time point to wake up at |
| 243 | + * @return Awaitable future that resumes at `until` |
220 | 244 | */ |
221 | 245 | TFuture<void> Sleep(TTime until); |
222 | 246 |
|
223 | 247 | /** |
224 | | - * @brief Sleep for a specific duration |
225 | | - * @tparam Rep Duration representation type |
| 248 | + * @brief Suspend the current handler for a duration |
| 249 | + * |
| 250 | + * Convenience overload for relative delays. Must be `co_await`-ed. |
| 251 | + * |
| 252 | + * @code |
| 253 | + * co_await ctx->Sleep(std::chrono::seconds(5)); |
| 254 | + * @endcode |
| 255 | + * |
| 256 | + * @tparam Rep Duration representation type |
226 | 257 | * @tparam Period Duration period type |
227 | 258 | * @param duration How long to sleep |
228 | | - * @return Future that completes when the sleep duration has elapsed |
| 259 | + * @return Awaitable future that resumes after `duration` |
229 | 260 | */ |
230 | 261 | template<typename Rep, typename Period> |
231 | 262 | TFuture<void> Sleep(std::chrono::duration<Rep,Period> duration); |
232 | 263 |
|
233 | 264 | /** |
234 | | - * @brief Send a message and wait for a response |
235 | | - * @tparam T Expected response message type |
| 265 | + * @brief Send a request and suspend until a reply of type `T` arrives |
| 266 | + * |
| 267 | + * Sends `question` to `recipient` and returns a future that resolves to the |
| 268 | + * first reply of type `T` sent back to this actor. Must be `co_await`-ed |
| 269 | + * inside `ICoroActor::CoReceive` or an async `TBehavior::Receive`. |
| 270 | + * |
| 271 | + * @code |
| 272 | + * auto reply = co_await ctx->Ask<Pong>(pingActor, Ping{}); |
| 273 | + * @endcode |
| 274 | + * |
| 275 | + * @tparam T Expected response message type |
236 | 276 | * @tparam TQuestion Question message type |
237 | 277 | * @param recipient Actor to send the question to |
238 | | - * @param question Question message to send |
239 | | - * @return Future containing the response message |
| 278 | + * @param question Message forwarded to the recipient |
| 279 | + * @return Awaitable future resolving to `T` when the reply arrives |
240 | 280 | */ |
241 | 281 | template<typename T, typename TQuestion> |
242 | 282 | TFuture<T> Ask(TActorId recipient, TQuestion&& question); |
@@ -332,30 +372,46 @@ class IActor { |
332 | 372 | /** |
333 | 373 | * @brief Coroutine-based actor interface for asynchronous message processing |
334 | 374 | * |
335 | | - * ICoroActor extends IActor to support asynchronous message processing |
336 | | - * using coroutines. The CoReceive method can perform async operations |
337 | | - * like sleeping, waiting for responses, or doing I/O without blocking |
338 | | - * the actor system. |
| 375 | + * Extends `IActor` so that `CoReceive` can `co_await` timers, `Ask`, or I/O |
| 376 | + * without blocking the event loop thread. |
| 377 | + * |
| 378 | + * **One async handler at a time.** When `CoReceive` returns a pending |
| 379 | + * `TFuture<void>`, the actor's mailbox is paused: no further messages are |
| 380 | + * dispatched until that future completes. This removes the need for internal |
| 381 | + * locking — state can be mutated freely inside the handler. |
| 382 | + * |
| 383 | + * @code |
| 384 | + * class Throttled : public ICoroActor { |
| 385 | + * TFuture<void> CoReceive(TMessageId id, TBlob blob, TActorContext::TPtr ctx) override { |
| 386 | + * auto msg = DeserializeNear<Work>(blob); |
| 387 | + * co_await ctx->Sleep(std::chrono::milliseconds(50)); // next msg waits here |
| 388 | + * process(msg); |
| 389 | + * } |
| 390 | + * }; |
| 391 | + * @endcode |
339 | 392 | */ |
340 | 393 | class ICoroActor : public IActor { |
341 | 394 | public: |
342 | 395 | /** |
343 | | - * @brief Synchronous receive method (calls CoReceive internally) |
344 | | - * @param messageId Type identifier of the message |
345 | | - * @param blob Serialized message data |
346 | | - * @param ctx Actor context for communication |
| 396 | + * @brief IActor bridge — invokes CoReceive and parks pending futures |
| 397 | + * |
| 398 | + * Called by the actor system. Not meant to be overridden; override |
| 399 | + * `CoReceive` instead. |
347 | 400 | */ |
348 | 401 | void Receive(TMessageId messageId, TBlob blob, TActorContext::TPtr ctx) override; |
349 | 402 |
|
350 | 403 | /** |
351 | | - * @brief Asynchronous message processing method |
352 | | - * @param messageId Type identifier of the message |
353 | | - * @param blob Serialized message data |
354 | | - * @param ctx Actor context for communication |
355 | | - * @return Future that completes when message processing is done |
| 404 | + * @brief Asynchronous message handler (override in subclass) |
| 405 | + * |
| 406 | + * May freely use `co_await ctx->Sleep(...)`, `co_await ctx->Ask<T>(...)`, |
| 407 | + * or any other coroutine primitive. While this coroutine is suspended, |
| 408 | + * the actor's mailbox is paused — subsequent messages queue up and are |
| 409 | + * delivered only after this handler returns (i.e. its future completes). |
356 | 410 | * |
357 | | - * This method can use co_await to perform asynchronous operations |
358 | | - * without blocking the actor system thread. |
| 411 | + * @param messageId Type identifier of the incoming message |
| 412 | + * @param blob Serialized message payload |
| 413 | + * @param ctx Actor context (valid for the duration of this handler) |
| 414 | + * @return Future that signals completion of message processing |
359 | 415 | */ |
360 | 416 | virtual TFuture<void> CoReceive(TMessageId messageId, TBlob blob, TActorContext::TPtr ctx) = 0; |
361 | 417 | }; |
@@ -389,22 +445,25 @@ class IBehavior { |
389 | 445 | * message handling. It supports both synchronous and asynchronous |
390 | 446 | * message handlers, automatically detecting the return type. |
391 | 447 | * |
| 448 | + * The handler return type determines dispatch mode automatically: |
| 449 | + * - `void` → synchronous, called inline |
| 450 | + * - `TFuture<void>` → asynchronous, mailbox pauses until the future completes |
| 451 | + * |
392 | 452 | * Usage example: |
393 | 453 | * @code |
394 | | - * class MyBehavior : public TBehavior<MyBehavior, MessageA, MessageB> { |
| 454 | + * class MyBehavior : public TBehavior<MyBehavior, Ping, Upload> { |
395 | 455 | * public: |
396 | | - * void Receive(MessageA&& msg, TBlob blob, TActorContext::TPtr ctx) { |
397 | | - * // Handle MessageA synchronously |
| 456 | + * void Receive(Ping&& msg, TBlob, TActorContext::TPtr ctx) { |
| 457 | + * ctx->Send<Pong>(ctx->Sender()); // synchronous reply |
398 | 458 | * } |
399 | 459 | * |
400 | | - * TFuture<void> Receive(MessageB&& msg, TBlob blob, TActorContext::TPtr ctx) { |
401 | | - * // Handle MessageB asynchronously |
402 | | - * co_await someAsyncOperation(); |
| 460 | + * TFuture<void> Receive(Upload&& msg, TBlob, TActorContext::TPtr ctx) { |
| 461 | + * auto sender = ctx->Sender(); |
| 462 | + * co_await ctx->Sleep(std::chrono::seconds(1)); // async — mailbox pauses |
| 463 | + * ctx->Send<UploadDone>(sender); |
403 | 464 | * } |
404 | 465 | * |
405 | | - * void HandleUnknownMessage(TMessageId messageId, TBlob blob, TActorContext::TPtr ctx) { |
406 | | - * // Handle messages not in the template parameter list |
407 | | - * } |
| 466 | + * void HandleUnknownMessage(TMessageId id, TBlob, TActorContext::TPtr) {} |
408 | 467 | * }; |
409 | 468 | * @endcode |
410 | 469 | */ |
@@ -489,44 +548,51 @@ class TBehavior : public IBehavior |
489 | 548 | /** |
490 | 549 | * @brief Actor that delegates message handling to a pluggable behavior |
491 | 550 | * |
492 | | - * IBehaviorActor allows actors to change their message handling behavior |
493 | | - * dynamically at runtime. This is useful for implementing state machines, |
494 | | - * protocol handlers, or any actor that needs to change its behavior based |
495 | | - * on its current state. |
| 551 | + * Enables runtime behavior switching — useful for state machines and protocol |
| 552 | + * handlers. Most commonly the actor itself is the initial behavior (`Become(this)`), |
| 553 | + * but separate `TBehavior` objects can be stored as members and swapped in. |
496 | 554 | * |
497 | 555 | * Usage example: |
498 | 556 | * @code |
499 | | - * class StatefulActor : public IBehaviorActor, |
500 | | - * public TBehavior<StatefulActor, InitMessage, WorkMessage> { |
| 557 | + * struct IdleBehavior; |
| 558 | + * struct ActiveBehavior; |
| 559 | + * |
| 560 | + * class Connection : public IBehaviorActor, |
| 561 | + * public TBehavior<Connection, Connect, Data, Disconnect> { |
501 | 562 | * public: |
502 | | - * StatefulActor() { |
503 | | - * Become(this); // Set initial behavior to self |
| 563 | + * Connection() { Become(this); } |
| 564 | + * |
| 565 | + * void Receive(Connect&&, TBlob, TActorContext::TPtr ctx) { |
| 566 | + * connected_ = true; |
| 567 | + * // next message will already use this actor as behavior (no switch needed) |
504 | 568 | * } |
505 | 569 | * |
506 | | - * void Receive(InitMessage&& msg, TBlob blob, TActorContext::TPtr ctx) { |
507 | | - * // Initialize and potentially switch to a different behavior |
508 | | - * if (msg.mode == "advanced") { |
509 | | - * Become(&advancedBehavior_); |
510 | | - * } |
| 570 | + * void Receive(Data&& msg, TBlob, TActorContext::TPtr ctx) { |
| 571 | + * if (!connected_) return; |
| 572 | + * process(msg.payload); |
511 | 573 | * } |
512 | 574 | * |
513 | | - * void HandleUnknownMessage(TMessageId messageId, TBlob blob, TActorContext::TPtr ctx) { |
514 | | - * // Handle unknown messages |
| 575 | + * void Receive(Disconnect&&, TBlob, TActorContext::TPtr) { |
| 576 | + * connected_ = false; |
515 | 577 | * } |
516 | 578 | * |
| 579 | + * void HandleUnknownMessage(TMessageId, TBlob, TActorContext::TPtr) {} |
| 580 | + * |
517 | 581 | * private: |
518 | | - * AdvancedBehavior advancedBehavior_; |
| 582 | + * bool connected_ = false; |
519 | 583 | * }; |
520 | 584 | * @endcode |
521 | 585 | */ |
522 | 586 | class IBehaviorActor : public IActor { |
523 | 587 | public: |
524 | 588 | /** |
525 | 589 | * @brief Switch to a new behavior |
526 | | - * @param behavior Pointer to the new behavior to use |
527 | 590 | * |
528 | | - * After calling Become(), all subsequent messages will be handled |
529 | | - * by the new behavior until Become() is called again. |
| 591 | + * The switch takes effect on the **next** incoming message. The behavior |
| 592 | + * pointer must remain valid for the lifetime of the actor — storing |
| 593 | + * behaviors as member variables is the typical pattern. |
| 594 | + * |
| 595 | + * @param behavior Non-owning pointer to the new behavior; must not be null |
530 | 596 | */ |
531 | 597 | void Become(IBehavior* behavior) { |
532 | 598 | CurrentBehavior_ = behavior; |
|
0 commit comments