mahler_core/runtime/
channel.rs

1use crate::error::Error;
2use crate::json::{Patch, Value};
3use crate::result::Result;
4use crate::sync::Sender;
5
6/// A channel to communicate state changes at runtime
7///
8/// The `Channel` allows tasks to send state changes and rollback checkpoints back to the worker during execution,
9/// enabling real-time progress updates. This can be used by extractors to propagate changes
10/// during a long operation.
11#[derive(Clone)]
12pub struct Channel {
13    sender: Option<Sender<(Patch, Option<Value>)>>,
14}
15
16impl std::fmt::Debug for Channel {
17    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
18        f.debug_struct("Channel")
19            .field(
20                "sender",
21                if self.sender.is_some() {
22                    &"attached"
23                } else {
24                    &"detached"
25                },
26            )
27            .finish()
28    }
29}
30
31impl Channel {
32    /// Return true if the channel is detached
33    pub fn is_detached(&self) -> bool {
34        self.sender.is_none()
35    }
36
37    /// Create a detached channel. A detached channel is not connected to a worker
38    pub fn detached() -> Self {
39        Self { sender: None }
40    }
41
42    /// Communicate the changes to the global state
43    pub async fn send(&self, changes: Patch) -> Result<()> {
44        if let Some(sender) = self.sender.as_ref() {
45            sender
46                .send((changes, None))
47                .await
48                .map_err(Error::internal)?;
49        }
50        Ok(())
51    }
52
53    /// Communicate changes and update the checkpoint
54    ///
55    /// Like [`send`](Channel::send), but also sends the given value as the
56    /// current checkpoint alongside the patch.
57    pub async fn send_and_commit(&self, changes: Patch, checkpoint: Value) -> Result<()> {
58        if let Some(sender) = self.sender.as_ref() {
59            sender
60                .send((changes, Some(checkpoint)))
61                .await
62                .map_err(Error::internal)?;
63        }
64        Ok(())
65    }
66}
67
68impl From<Sender<(Patch, Option<Value>)>> for Channel {
69    /// Create an attached channel from a sender.
70    fn from(sender: Sender<(Patch, Option<Value>)>) -> Self {
71        Self {
72            sender: Some(sender),
73        }
74    }
75}
mahler_core/runtime/channel.rs

mahler_core/runtime/
channel.rs
