From 2e1f9261f0cada7ecadcc731313d46ef27addf16 Mon Sep 17 00:00:00 2001 From: Jeremy HERGAULT Date: Fri, 17 Jul 2026 15:48:08 +0200 Subject: [PATCH 1/2] feat: implement APN Signed-off-by: Jeremy HERGAULT --- prosa/src/core.rs | 2 + prosa/src/core/apn.rs | 459 ++++++++++++++++++++++++++++++++++ prosa_book/src/SUMMARY.md | 1 + prosa_book/src/ch03-10-apn.md | 121 +++++++++ 4 files changed, 583 insertions(+) create mode 100644 prosa/src/core/apn.rs create mode 100644 prosa_book/src/ch03-10-apn.md diff --git a/prosa/src/core.rs b/prosa/src/core.rs index 428dc69..219d124 100644 --- a/prosa/src/core.rs +++ b/prosa/src/core.rs @@ -24,6 +24,8 @@ /// Adaptor module to adapt processor object and internal messages pub mod adaptor; +/// APN (Application Programming Node) module to run inline service-call automatons from a processor +pub mod apn; /// Define error types for adaptor and processor pub mod error; /// The module define ProSA main processing to bring asynchronous handler for all processors diff --git a/prosa/src/core/apn.rs b/prosa/src/core/apn.rs new file mode 100644 index 0000000..2883523 --- /dev/null +++ b/prosa/src/core/apn.rs @@ -0,0 +1,459 @@ +//! APN (Application Programming Node): service-call automatons. +//! +//! An APN lets a processor react to a service request by running an *automaton* that can sub-call +//! other services, branch on their results, and produce the final response — without writing a +//! manual state machine over the processor loop. +//! +//! An APN is launched with [`RequestMsg::apn`]: it spawns the automaton on a Tokio task, handing it +//! an [`Apn`] handle plus the request's service name and data. The automaton issues sub-calls with +//! [`Apn::call`], branches on their results, and returns the final `M`; the APN sends that result back +//! to the original caller on the request's response queue. The processor loop is **not blocked**. +//! +//! Because the automaton is spawned, it must be `Send + 'static`: it captures owned data and cannot +//! borrow the processor's state. +//! +//! An APN only processes service requests: the whole automaton runs under a timeout budget, and an +//! APN should never open sockets, wait on timers, or block for a long time. If you need any of those, +//! write a full ProSA processor instead. + +use std::sync::Arc; +use std::time::Duration; + +use tokio::spawn; +use tokio::sync::{mpsc, oneshot}; +use tokio::time::Instant; + +use super::{ + msg::{InternalMsg, Msg, RequestMsg, ResponseMsg, Tvf}, + service::{ServiceError, ServiceTable}, +}; + +/// Handle used by an APN automaton to issue sub-calls to other services. +/// +/// Each sub-call gets its own oneshot response channel, so a reply can never be mistaken for another +/// call's — no shared queue, no correlation needed. Sub-calls propagate the original request's trace +/// span, so their traces are nested under it. +/// +/// An APN is launched with [`RequestMsg::apn`]. +pub struct Apn +where + M: Sized + Clone + Tvf, +{ + service_table: Arc>, + timeout: Duration, + trace_id: Option, +} + +impl std::fmt::Debug for Apn +where + M: Sized + Clone + Tvf, +{ + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Apn") + .field("timeout", &self.timeout) + .field("trace_id", &self.trace_id) + .finish() + } +} + +impl Apn +where + M: Sized + Clone + Tvf, +{ + /// Create an APN from a service table snapshot and the overall automaton timeout budget. + /// + /// Usually you don't call this directly — use [`RequestMsg::apn`], which builds the APN and + /// launches its automaton on a spawned task. + pub(crate) fn new( + service_table: Arc>, + timeout: Duration, + trace_id: Option, + ) -> Apn { + Apn { + service_table, + timeout, + trace_id, + } + } + + /// The trace span id this APN propagates to its sub-calls, or `None` if the original request + /// carried no span. Sub-calls set it automatically; use this only if the automaton needs to build + /// its own spans nested under the request. + pub fn trace_id(&self) -> Option<&tracing::span::Id> { + self.trace_id.as_ref() + } + + /// Sub-call a service and await its response. + /// + /// This call is not individually timed out: it is bounded only by the APN's overall budget (the + /// `timeout` given to [`RequestMsg::apn`], which caps the whole automaton). Use + /// [`Apn::call_with_timeout`] to bound a single sub-call. + /// + /// It borrows `&self`, so several sub-calls can be issued concurrently (e.g. with + /// [`tokio::join!`]); each one has its own response channel, so their replies never interfere. + /// + /// Returns the response on success, or a [`ServiceError`] if the service can't be reached or + /// returns an error. + pub async fn call(&self, service_name: &str, data: M) -> Result, ServiceError> { + let Some(proc_queue) = self + .service_table + .get_proc_service(service_name) + .map(|proc_service| proc_service.proc_queue.clone()) + else { + return Err(ServiceError::UnableToReachService(service_name.to_string())); + }; + self.dispatch(&proc_queue, service_name, data, None).await + } + + /// Sub-call a service and await its response, bounding this single call with an explicit timeout. + /// + /// Returns the response on success, or a [`ServiceError`] if the service can't be reached, + /// doesn't respond within `timeout`, or returns an error. + pub async fn call_with_timeout( + &self, + service_name: &str, + data: M, + timeout: Duration, + ) -> Result, ServiceError> { + let Some(proc_queue) = self + .service_table + .get_proc_service(service_name) + .map(|proc_service| proc_service.proc_queue.clone()) + else { + return Err(ServiceError::UnableToReachService(service_name.to_string())); + }; + self.dispatch(&proc_queue, service_name, data, Some(timeout)) + .await + } + + /// Send a request to a processor queue and await its response on a dedicated oneshot channel. + async fn dispatch( + &self, + proc_queue: &mpsc::Sender>, + service_name: &str, + data: M, + timeout: Option, + ) -> Result, ServiceError> { + let (response_tx, response_rx) = oneshot::channel(); + let request = match &self.trace_id { + Some(trace_id) => RequestMsg::new_with_trace_id( + service_name.to_string(), + data, + response_tx, + trace_id.clone(), + ), + None => RequestMsg::new(service_name.to_string(), data, response_tx), + }; + + if proc_queue + .send(InternalMsg::Request(request)) + .await + .is_err() + { + return Err(ServiceError::UnableToReachService(service_name.to_string())); + } + + if let Some(timeout) = timeout { + match tokio::time::timeout(timeout, response_rx).await { + Ok(Ok(InternalMsg::Response(resp))) => Ok(resp), + Ok(Ok(InternalMsg::Error(err))) => Err(err.into_err()), + Ok(Ok(_)) => Err(ServiceError::ProtocolError(service_name.to_string())), + Ok(Err(_recv)) => Err(ServiceError::UnableToReachService(service_name.to_string())), + Err(_elapsed) => Err(ServiceError::Timeout( + service_name.to_string(), + timeout.as_millis() as u64, + )), + } + } else { + match response_rx.await { + Ok(InternalMsg::Response(resp)) => Ok(resp), + Ok(InternalMsg::Error(err)) => Err(err.into_err()), + Ok(_) => Err(ServiceError::ProtocolError(service_name.to_string())), + Err(_recv) => Err(ServiceError::UnableToReachService(service_name.to_string())), + } + } + } +} + +impl RequestMsg +where + M: Sized + + Clone + + std::fmt::Debug + + Tvf + + Default + + 'static + + std::marker::Send + + std::marker::Sync, +{ + /// Run an APN automaton for this request on a spawned Tokio task. + /// + /// The automaton is handed an [`Apn`] handle plus this request's **service name and data**. It + /// issues sub-calls with [`Apn::call`], branches on their results, and returns the final `M`. That + /// result is sent back to the original requestor on this request's response queue — you never call + /// `return_to_sender` yourself. The request's trace span is available to the automaton through + /// [`Apn::trace_id`]. + /// + /// The automaton runs on its own task, so the processor loop is not blocked. Because it is + /// spawned, the closure must be `Send + 'static`: it captures owned data only and cannot borrow + /// the processor's state. The `timeout` is the overall budget for the whole automaton; keep it + /// short. + /// + /// ```no_run + /// use std::sync::Arc; + /// use std::time::Duration; + /// use prosa::core::msg::{Msg, RequestMsg, Tvf}; + /// use prosa::core::service::ServiceTable; + /// use prosa_utils::msg::simple_string_tvf::SimpleStringTvf; + /// + /// fn handle( + /// request: RequestMsg, + /// services: Arc>, + /// ) { + /// request.apn( + /// services.clone(), + /// Duration::from_millis(500), + /// move |apn, _service, data| async move { + /// let mut resp = match data.get_unsigned(1).unwrap_or(0) { + /// 0 => apn.call("PAY", data).await?, + /// _ => apn.call("REJECT", data).await?, + /// }; + /// Ok(resp.take_data().unwrap_or_default()) + /// }, + /// ); + /// } + /// ``` + /// + /// Sub-calls can also run concurrently: [`Apn::call`] borrows `&self`, so fan out to distinct + /// services with [`tokio::join!`] and await them together: + /// + /// ```no_run + /// use std::sync::Arc; + /// use std::time::Duration; + /// use prosa::core::msg::{Msg, RequestMsg, Tvf}; + /// use prosa::core::service::ServiceTable; + /// use prosa_utils::msg::simple_string_tvf::SimpleStringTvf; + /// + /// fn handle( + /// request: RequestMsg, + /// services: Arc>, + /// ) { + /// request.apn( + /// services.clone(), + /// Duration::from_millis(500), + /// move |apn, _service, data| async move { + /// // Fire both sub-calls, then await both. + /// let (pay, fraud) = tokio::join!( + /// apn.call("PAY", data.clone()), + /// apn.call("FRAUD", data), + /// ); + /// let mut pay = pay?; + /// let _fraud = fraud?; + /// Ok(pay.take_data().unwrap_or_default()) + /// }, + /// ); + /// } + /// ``` + pub fn apn( + mut self, + service_table: Arc>, + timeout: Duration, + automaton: F, + ) where + F: FnOnce(Apn, String, M) -> Fut + Send + 'static, + Fut: Future> + Send + 'static, + { + let apn = Apn::new(service_table, timeout, self.get_span().id()); + let service = self.get_service().clone(); + let data = self.take_data().unwrap_or_default(); + + let deadline = Instant::now() + timeout; + spawn(async move { + let _ = match tokio::time::timeout_at(deadline, automaton(apn, service, data)).await { + Ok(result) => self.return_result_to_sender(result), + Err(_elapsed) => { + let service_name = self.get_service().to_string(); + self.return_error_to_sender( + None, + ServiceError::Timeout(service_name, timeout.as_millis() as u64), + ) + } + }; + }); + } +} + +#[cfg(test)] +mod tests { + extern crate self as prosa; + + use std::sync::Arc; + use std::time::Duration; + + use prosa_macros::{proc, settings}; + use prosa_utils::msg::{simple_string_tvf::SimpleStringTvf, tvf::Tvf}; + use serde::Serialize; + use tokio::sync::mpsc; + use tokio::time::timeout; + + use super::Apn; + use crate::core::{ + error::BusError, + main::{Main, MainProc, MainRunnable}, + msg::{InternalMsg, Msg, RequestMsg}, + proc::{ProcBusParam, ProcConfig, ProcParam}, + service::{ProcService, ServiceError, ServiceTable}, + }; + use crate::stub::adaptor::StubParotAdaptor; + use crate::stub::proc::{StubProc, StubSettings}; + + /// Dummy settings for building throwaway `Main` handles in unit tests + #[settings] + #[derive(Default, Debug, Serialize)] + struct DummySettings {} + + #[tokio::test] + async fn apn_call_unreachable_service() { + let apn: Apn = Apn::new( + Arc::new(ServiceTable::default()), + Duration::from_millis(50), + None, + ); + let err = apn + .call("NOPE", SimpleStringTvf::default()) + .await + .expect_err("unreachable service should error"); + assert!(matches!(err, ServiceError::UnableToReachService(_))); + } + + #[tokio::test] + async fn apn_call_timeout() { + // Build a service table pointing at a queue that no one ever reads: the request is + // buffered but never answered, so the sub-call must time out. + let (bus, _main): (Main, MainProc) = + MainProc::create(&DummySettings::default(), None); + let (queue_tx, _queue_rx) = mpsc::channel(8); + let proc_param = ProcParam::new(1, "slow".to_string(), queue_tx.clone(), bus); + let proc_service = ProcService::new(&proc_param, queue_tx, 0); + + let mut table = ServiceTable::default(); + table.add_service("SLOW", proc_service); + + let apn: Apn = Apn::new(Arc::new(table), Duration::from_millis(30), None); + let err = apn + .call_with_timeout( + "SLOW", + SimpleStringTvf::default(), + Duration::from_millis(30), + ) + .await + .expect_err("slow service should time out"); + assert!(matches!(err, ServiceError::Timeout(_, _))); + + // Keep the never-read receiver alive until the assertion is done. + drop(_queue_rx); + } + + #[proc] + struct ApnTestProc {} + + #[proc] + impl ApnTestProc { + async fn apn_run(&mut self) -> Result<(), BusError> { + self.proc.add_proc().await?; + self.proc + .add_service_proc(vec![String::from("APN")]) + .await?; + + let mut sent = false; + loop { + if let Some(msg) = self.internal_rx_queue.recv().await { + match msg { + InternalMsg::Service(table) => { + self.service = table; + if !sent + && self.service.exist_proc_service("SUB1") + && self.service.exist_proc_service("SUB2") + && self.service.exist_proc_service("APN") + { + sent = true; + let mut data = SimpleStringTvf::default(); + data.put_string(1, "start"); + if let Some(service) = self.service.get_proc_service("APN") { + service + .proc_queue + .send(InternalMsg::Request(RequestMsg::new( + String::from("APN"), + data, + self.proc.get_service_queue(), + ))) + .await + .expect("APN request should be sent"); + } + } + } + InternalMsg::Request(req) => { + // Run an APN for the request: the automaton chains SUB1 then SUB2. + req.apn( + self.service.clone(), + Duration::from_millis(500), + move |apn, _service, data| async move { + let mut first = apn.call("SUB1", data).await?; + let first_data = first.take_data().ok_or_else(|| { + ServiceError::ProtocolError("SUB1".to_string()) + })?; + let mut resp = apn.call("SUB2", first_data).await?; + resp.take_data().ok_or_else(|| { + ServiceError::ProtocolError("SUB2".to_string()) + }) + }, + ); + } + InternalMsg::Response(resp) => { + assert_eq!("start", resp.get_data()?.get_string(1)?.into_owned()); + self.proc.remove_proc(None).await?; + return Ok(()); + } + InternalMsg::Error(err) => { + return Err(BusError::ProcComm( + self.get_proc_id(), + 0, + format!("unexpected APN error: {:?}", err.get_err()), + )); + } + _ => {} + } + } + } + } + } + + #[tokio::test] + async fn apn_happy_path() { + let (bus, main) = MainProc::::create(&DummySettings::default(), Some(2)); + let main_task = tokio::spawn(main.run()); + + // Stub offering the two sub-services the automaton chains through (parrot echoes data). + let stub_proc = StubProc::::create( + 1, + String::from("stub"), + bus.clone(), + StubSettings::new(vec![String::from("SUB1"), String::from("SUB2")]), + ); + crate::core::proc::Proc::::run(stub_proc).expect("stub should run"); + + let result = timeout( + Duration::from_secs(5), + ApnTestProc::::create_raw(2, "apn_test".to_string(), bus.clone()) + .apn_run(), + ) + .await + .expect("APN test should not time out"); + assert_eq!(Ok(()), result); + + bus.stop("ProSA unit test end".into()) + .await + .expect("ProSA should stop"); + main_task.await.expect("Main task should end correctly"); + } +} diff --git a/prosa_book/src/SUMMARY.md b/prosa_book/src/SUMMARY.md index 801b2d4..2fb5b5e 100644 --- a/prosa_book/src/SUMMARY.md +++ b/prosa_book/src/SUMMARY.md @@ -35,3 +35,4 @@ - [I/O](ch03-07-io.md) - [Threads](ch03-08-threads.md) - [Built-in Processors](ch03-09-builtin.md) + - [APN](ch03-10-apn.md) diff --git a/prosa_book/src/ch03-10-apn.md b/prosa_book/src/ch03-10-apn.md new file mode 100644 index 0000000..6109253 --- /dev/null +++ b/prosa_book/src/ch03-10-apn.md @@ -0,0 +1,121 @@ +# APN + +An **APN** (Application Programming Node) lets a processor react to a service request by running a small *automaton* that can call other services, branch on their results, and produce the final response — without hand-writing a state machine over the processor loop. + +An APN associates a service request with an automaton that handles it. Your code implements the automaton, while the framework takes care of running it and returning the response. ProSA exposes this as a lightweight primitive rather than a dedicated processor: you launch an APN directly on a request from within any processor. + +## When to use an APN + +Use an APN when handling a request means *making one or more sub-calls and deciding what to do next based on their responses*. For example: call an authorization service, and depending on its return code, call either a payment service or a rejection service, then return the outcome to the original caller. + +Without an APN you would have to store the in-flight request in a [`PendingMsgs`](./ch03-06-events.md) map, send the sub-call yourself, match the response on a later loop iteration, correlate it back, and repeat for every step. An APN collapses all of that into a single linear (or branching) block of `async` code. + +## Limitations + +An APN has a few limitations by design: + +- **It only processes service requests.** It cannot drive a timer, open a socket, or manage any external resource. +- **The automaton runs under a timeout budget.** An APN must never block for long; keep the budget short. + +The automaton runs on its own spawned task, so it does **not** block the processor loop — but because it is spawned it must be `Send + 'static`: it captures owned data only and cannot borrow the processor's state. + +If your need matches any of these limitations, write a full ProSA [processor](./ch03-00-proc.md) instead. + +## Usage + +An APN is launched with [`RequestMsg::apn`](https://docs.rs/prosa/latest/prosa/core/msg/struct.RequestMsg.html#method.apn), called directly on the request you want to handle. It takes a service table snapshot and a timeout that is the **overall budget for the whole automaton**, spawns a Tokio task for the automaton, and returns straight away — it is a plain (non-`async`) call. + +The APN hands the automaton closure the [`Apn`](https://docs.rs/prosa/latest/prosa/core/apn/struct.Apn.html) handle plus the request's **service name and data** (a `String` and the `M`). There is no automatic first call: the automaton drives every sub-call itself with `apn.call(...)`, branches on the results, and returns the final `M`. That result is **sent back** to the original requestor on the request's response queue — you never call `return_to_sender` yourself. If the automaton needs the request's trace span (to nest its own spans), it's available via [`apn.trace_id()`](https://docs.rs/prosa/latest/prosa/core/apn/struct.Apn.html#method.trace_id). + +Build a [`RequestMsg`](https://docs.rs/prosa/latest/prosa/core/msg/struct.RequestMsg.html) with its **response queue** set to your processor's own service queue ([`get_service_queue()`](https://docs.rs/prosa/latest/prosa/core/proc/struct.ProcParam.html#method.get_service_queue)) — that queue is where the APN's final result lands — then call [`apn`](https://docs.rs/prosa/latest/prosa/core/msg/struct.RequestMsg.html#method.apn) on it. Since the automaton is spawned, everything it needs must be captured by value: + +```rust,noplayground +// `trans` is the request to handle (its response queue points back at this processor). +trans.apn( + self.service.clone(), + self.settings.apn_timeout, + move |apn, _service, data| async move { + // `data` is the request payload; drive the sub-calls from it. + let mut auth = apn.call("AUTH", data).await?; + let auth_data = auth.take_data().unwrap_or_default(); + let mut resp = match auth_data.get_unsigned(1).unwrap_or(0) { + 0 => apn.call("PAY", auth_data).await?, // final response, auto-sent + _ => apn.call("REJECT", auth_data).await?, + }; + Ok(resp.take_data().unwrap_or_default()) + }, +); +``` + +The automaton can create any object it needs; just remember it captures owned values (clone what you need out of the processor before launching). + +### Sub-calls + +The [`Apn`](https://docs.rs/prosa/latest/prosa/core/apn/struct.Apn.html) handle exposes two methods: + +- [`call()`](https://docs.rs/prosa/latest/prosa/core/apn/struct.Apn.html#method.call) — sub-call a service; not individually timed out, it is bounded only by the APN's overall timeout budget. +- [`call_with_timeout()`](https://docs.rs/prosa/latest/prosa/core/apn/struct.Apn.html#method.call_with_timeout) — sub-call with an explicit timeout for that one call. + +Both return `Result, ServiceError>` — take the data out of the response to use it — so failures are handled with ordinary `?` / `match`: + +- `ServiceError::UnableToReachService` — the service isn't in the table, or the send failed. +- `ServiceError::Timeout` — the service didn't respond within the timeout. +- Any error returned by the sub-called service is forwarded as-is. + +Each sub-call gets its own dedicated response channel, so a reply can never be mistaken for another call's. Sub-call traces are nested under the original request's span, so a full APN flow shows up as a single trace tree. + +If a sub-call fails (unreachable or timeout), propagate it with `?` and the error is returned straight to the original caller. + +### Parallel sub-calls + +Because [`call()`](https://docs.rs/prosa/latest/prosa/core/apn/struct.Apn.html#method.call) borrows `&self`, an automaton can fan out to several distinct services at once and await them together with [`tokio::join!`](https://docs.rs/tokio/latest/tokio/macro.join.html) — each sub-call has its own response channel, so their replies never interfere: + +```rust,noplayground +move |apn, _service, data| async move { + // Fire both sub-calls, then await both. + let (pay, fraud) = tokio::join!( + apn.call("PAY", data.clone()), + apn.call("FRAUD", data), + ); + let mut pay = pay?; + let _fraud = fraud?; + Ok(pay.take_data().unwrap_or_default()) +} +``` + +## States + +The "each state is a separate implementation" model maps naturally onto plain control flow: the automaton *is* the closure, and it holds its own state. A multi-state machine is just a loop over an enum, where each arm may issue a sub-call and transition to the next state. The request data seeds the initial state: + +```rust,noplayground +enum State { + Start(M), + Authorized(M), + Paid(M), +} + +trans.apn(self.service.clone(), timeout, move |apn, _service, data| async move { + let mut state = State::Start(data); + loop { + state = match state { + State::Start(s) => { + State::Authorized(apn.call("AUTH", s).await?.take_data().unwrap_or_default()) + } + State::Authorized(a) if a.get_unsigned(1).unwrap_or(0) == 0 => { + State::Paid(apn.call("PAY", a).await?.take_data().unwrap_or_default()) + } + State::Authorized(a) => { + return Ok(apn.call("REJECT", a).await?.take_data().unwrap_or_default()); + } + State::Paid(p) => return Ok(p), + }; + } +}); +``` + +## Relation to `PendingMsgs` + +An APN and [`PendingMsgs`](./ch03-06-events.md) solve related problems from opposite ends: + +- Use an **APN** when the follow-up logic is a self-contained linear or branching flow that can run on its own task from captured data. +- Use **`PendingMsgs`** when you need the processor loop itself to keep driving each response and timeout across loop iterations — for instance to share and mutate processor state per response. From 93bd2a5fe61f1dd265d4ffc40e72d2aec44145d6 Mon Sep 17 00:00:00 2001 From: Jeremy HERGAULT Date: Fri, 17 Jul 2026 16:12:04 +0200 Subject: [PATCH 2/2] fix: doc Signed-off-by: Jeremy HERGAULT --- prosa/src/core/apn.rs | 9 +++++---- prosa_book/src/ch03-10-apn.md | 19 ++++++++++++++++++- 2 files changed, 23 insertions(+), 5 deletions(-) diff --git a/prosa/src/core/apn.rs b/prosa/src/core/apn.rs index 2883523..c173f4c 100644 --- a/prosa/src/core/apn.rs +++ b/prosa/src/core/apn.rs @@ -4,10 +4,11 @@ //! other services, branch on their results, and produce the final response — without writing a //! manual state machine over the processor loop. //! -//! An APN is launched with [`RequestMsg::apn`]: it spawns the automaton on a Tokio task, handing it -//! an [`Apn`] handle plus the request's service name and data. The automaton issues sub-calls with -//! [`Apn::call`], branches on their results, and returns the final `M`; the APN sends that result back -//! to the original caller on the request's response queue. The processor loop is **not blocked**. +//! An APN is launched with [`RequestMsg::apn`](crate::core::msg::RequestMsg::apn): it spawns the +//! automaton on a Tokio task, handing it an [`Apn`](crate::core::apn::Apn) handle plus the request's +//! service name and data. The automaton issues sub-calls with [`Apn::call`](crate::core::apn::Apn::call), +//! branches on their results, and returns the final `M`; the APN sends that result back to the original +//! caller on the request's response queue. The processor loop is **not blocked**. //! //! Because the automaton is spawned, it must be `Send + 'static`: it captures owned data and cannot //! borrow the processor's state. diff --git a/prosa_book/src/ch03-10-apn.md b/prosa_book/src/ch03-10-apn.md index 6109253..e5f9d72 100644 --- a/prosa_book/src/ch03-10-apn.md +++ b/prosa_book/src/ch03-10-apn.md @@ -27,7 +27,7 @@ An APN is launched with [`RequestMsg::apn`](https://docs.rs/prosa/latest/prosa/c The APN hands the automaton closure the [`Apn`](https://docs.rs/prosa/latest/prosa/core/apn/struct.Apn.html) handle plus the request's **service name and data** (a `String` and the `M`). There is no automatic first call: the automaton drives every sub-call itself with `apn.call(...)`, branches on the results, and returns the final `M`. That result is **sent back** to the original requestor on the request's response queue — you never call `return_to_sender` yourself. If the automaton needs the request's trace span (to nest its own spans), it's available via [`apn.trace_id()`](https://docs.rs/prosa/latest/prosa/core/apn/struct.Apn.html#method.trace_id). -Build a [`RequestMsg`](https://docs.rs/prosa/latest/prosa/core/msg/struct.RequestMsg.html) with its **response queue** set to your processor's own service queue ([`get_service_queue()`](https://docs.rs/prosa/latest/prosa/core/proc/struct.ProcParam.html#method.get_service_queue)) — that queue is where the APN's final result lands — then call [`apn`](https://docs.rs/prosa/latest/prosa/core/msg/struct.RequestMsg.html#method.apn) on it. Since the automaton is spawned, everything it needs must be captured by value: +You build that [`RequestMsg`](https://docs.rs/prosa/latest/prosa/core/msg/struct.RequestMsg.html) just as you would to [send a message to a service](./ch03-05-service.md#sending-messages) — with one twist: set its **response queue** to your processor's own service queue ([`get_service_queue()`](https://docs.rs/prosa/latest/prosa/core/proc/struct.ProcParam.html#method.get_service_queue)), so the APN's final result lands back in your own loop. Then, instead of pushing it onto a service's `proc_queue` yourself, hand it to [`apn`](https://docs.rs/prosa/latest/prosa/core/msg/struct.RequestMsg.html#method.apn), which takes ownership of the request from there. Since the automaton is spawned, everything it needs must be captured by value: ```rust,noplayground // `trans` is the request to handle (its response queue points back at this processor). @@ -47,6 +47,23 @@ trans.apn( ); ``` +Just as often you won't build a request at all: you'll run an APN on one you **received**. When your processor [offers a service](./ch03-05-service.md#listening-to-a-service), a [`RequestMsg`](https://docs.rs/prosa/latest/prosa/core/msg/struct.RequestMsg.html) arrives in the `Request` arm of its loop already carrying the response queue of whoever called you. Hand that request straight to [`apn`](https://docs.rs/prosa/latest/prosa/core/msg/struct.RequestMsg.html#method.apn) — no rebuilding, no re-wiring — and the automaton's result flows back to the original caller untouched: + +```rust,noplayground +InternalMsg::Request(request) => { + // A request for a service this processor offers just arrived. + request.apn( + self.service.clone(), + self.settings.apn_timeout, + move |apn, _service, data| async move { + // Drive the sub-calls from the received request's data. + let mut resp = apn.call("NEXT", data).await?; + Ok(resp.take_data().unwrap_or_default()) + }, + ); +} +``` + The automaton can create any object it needs; just remember it captures owned values (clone what you need out of the processor before launching). ### Sub-calls