dora_message/
daemon_to_daemon.rs

1use aligned_vec::{AVec, ConstAlign};
2
3use crate::{
4    DataflowId,
5    id::{DataId, NodeId},
6    metadata::Metadata,
7};
8
9#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
10#[allow(clippy::large_enum_variant)]
11#[non_exhaustive]
12pub enum InterDaemonEvent {
13    Output {
14        dataflow_id: DataflowId,
15        node_id: NodeId,
16        output_id: DataId,
17        metadata: Metadata,
18        #[serde(with = "crate::bulk_bytes::option")]
19        data: Option<AVec<u8, ConstAlign<128>>>,
20    },
21    OutputClosed {
22        dataflow_id: DataflowId,
23        node_id: NodeId,
24        output_id: DataId,
25    },
26    /// An opaque message between the daemons of one dataflow, sent on
27    /// behalf of an out-of-tree extension. dora never interprets
28    /// `namespace` or `payload` — it moves the bytes and nothing else.
29    ///
30    /// This is the inter-daemon half of the extension seam (the node-facing
31    /// halves are [`crate::node_to_daemon::DaemonRequest::ExtensionRequest`]
32    /// and the extension table). Naming the variants after any one
33    /// transport would freeze that transport's architecture into the 1.0
34    /// protocol, which is exactly what `docs/extensions.md` exists to
35    /// avoid; a second extension needs no change here at all.
36    ExtensionMessage {
37        dataflow_id: DataflowId,
38        /// Extension that owns the payload. A daemon with no extension
39        /// registered under this namespace drops the message.
40        namespace: String,
41        /// When set, only the daemon whose machine id matches acts on the
42        /// message; the others drop it. `None` addresses every daemon in
43        /// the dataflow.
44        target_machine: Option<String>,
45        #[serde(with = "crate::bulk_bytes::vec")]
46        payload: Vec<u8>,
47    },
48}
49
50impl InterDaemonEvent {
51    /// Bulk bytes this event will contribute to its encoding, for
52    /// [`crate::encode_presized`].
53    ///
54    /// Not to be confused with [`crate::metadata::debug_frame_wire_size`], which
55    /// answers "how big was this on the wire" and deliberately prefers the
56    /// daemon-stamped `WIRE_SIZE` parameter over the buffer length (#2584). This
57    /// one must be the actual buffer length, since it sizes an allocation.
58    pub fn encode_size_hint(&self) -> usize {
59        match self {
60            Self::Output { data, .. } => data.as_ref().map_or(0, |d| d.len()),
61            Self::OutputClosed { .. } => 0,
62            // Opaque to dora, but it can carry a full tensor frame.
63            Self::ExtensionMessage { payload, .. } => payload.len(),
64        }
65    }
66}
dora_message/daemon_to_daemon.rs

dora_message/
daemon_to_daemon.rs
