tocat_api/
lib.rs

1//! Public plugin API for tocat.
2//!
3//! A plugin is a synchronous byte transformer. It is handed a chunk of bytes
4//! that arrived from upstream and decides what to forward downstream. Anything
5//! that touches the outside world (writing a dump file, emitting a log line)
6//! is *not* performed by the plugin. It is queued as an [`Effect`] and applied
7//! by the host after the call returns.
8//!
9//! That split is deliberate. It keeps plugins pure and trivially testable, it
10//! keeps all I/O on the host's async runtime, and it is the shape a WASM guest
11//! has to take anyway (guest calls a host import, host performs the syscall).
12//! A future `WasmPlugin` implements [`Plugin`] like any other; nothing in the
13//! relay needs to change.
14//!
15//! The same split covers time. A stage cannot await and cannot read a clock (a
16//! guest has no way to reach one) so a stage that needs time rather than
17//! traffic to drive it declares a period with [`Plugin::tick_interval`] and is
18//! called back through [`Plugin::on_tick`]. The host holds the timer and
19//! decides when anyone is due.
20//!
21//! # Composition
22//!
23//! Plugins are declared once and instantiated per direction. A declaration list
24//! `[a, b]` with `direction = "both"` produces:
25//!
26//! ```text
27//! source --> a --> b --> sink        (Direction::SourceToSink)
28//! source <-- a <-- b <-- sink        (Direction::SinkToSource)
29//! ```
30//!
31//! The reverse pipeline is the *mirror* of the declaration order, so wrapping
32//! plugins (framing, compression, encryption) nest correctly without the user
33//! having to write the pipeline out twice. Each direction gets its own
34//! instance, so per-direction state (byte offsets, codec state) never leaks
35//! across paths.
36
37pub mod channel;
38pub mod error;
39pub mod forgiving;
40pub mod interval;
41pub mod normalize;
42pub mod pipeline;
43pub mod plugin;
44pub mod size;
45
46use std::{fmt, str::FromStr};
47
48use serde::{Deserialize, Deserializer, Serialize};
49use serde_json::{Map, Value};
50
51pub use crate::{
52    channel::{ChannelId, ChannelTarget, HostBuilder},
53    error::{PluginError, Result},
54    forgiving::Forgiving,
55    interval::{Interval, ParseIntervalError},
56    normalize::{canonical, normalize},
57    pipeline::{BoundaryFault, Chain, Emitted, Pipeline, Registry, Segment, Side},
58    plugin::{
59        Boundaries, BuildCtx, Ctx, EffectSink, Emission, Emit, Execution, ExternalStage, LogLevel,
60        Needs, PipelineMeta, Plugin, PluginFactory, Stage, StageInfo, StderrMode,
61    },
62    size::{ByteSize, ParseSizeError},
63};
64
65/// One of the two byte paths through the relay.
66#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
67#[serde(rename_all = "kebab-case")]
68pub enum Direction {
69    /// Bytes read from the source, written to the sink.
70    #[serde(alias = "forward", alias = "src-to-sink", alias = "source-to-sink")]
71    SourceToSink,
72    /// Bytes read from the sink, written to the source.
73    #[serde(alias = "reverse", alias = "sink-to-src", alias = "sink-to-source")]
74    SinkToSource,
75}
76
77impl Direction {
78    pub const ALL: [Direction; 2] = [Direction::SourceToSink, Direction::SinkToSource];
79
80    #[must_use]
81    pub fn flip(self) -> Self {
82        match self {
83            Direction::SourceToSink => Direction::SinkToSource,
84            Direction::SinkToSource => Direction::SourceToSink,
85        }
86    }
87
88    #[must_use]
89    pub fn as_str(self) -> &'static str {
90        match self {
91            Direction::SourceToSink => "source-to-sink",
92            Direction::SinkToSource => "sink-to-source",
93        }
94    }
95}
96
97impl fmt::Display for Direction {
98    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
99        f.write_str(self.as_str())
100    }
101}
102
103/// Which path(s) a declared plugin applies to.
104///
105/// The default is [`DirectionSpec::SourceToSink`], so an entry with no
106/// direction is on the forward path only. [`DirectionSpec::Both`] means that,
107/// on request: the plugin is instantiated twice, once per direction, rather
108/// than sharing one instance, and the sink-to-source chain is the mirror of the
109/// declaration order so that wrapping stages nest correctly.
110#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash, Serialize)]
111#[serde(rename_all = "kebab-case")]
112pub enum DirectionSpec {
113    #[default]
114    SourceToSink,
115    SinkToSource,
116    Both,
117}
118
119/// Deserialized through [`FromStr`] so that a config file accepts exactly what
120/// the command line does.
121impl<'de> Deserialize<'de> for DirectionSpec {
122    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> std::result::Result<Self, D::Error> {
123        let raw = String::deserialize(deserializer)?;
124        raw.parse().map_err(serde::de::Error::custom)
125    }
126}
127
128impl DirectionSpec {
129    #[must_use]
130    pub fn contains(self, direction: Direction) -> bool {
131        matches!(
132            (self, direction),
133            (DirectionSpec::Both, _)
134                | (DirectionSpec::SourceToSink, Direction::SourceToSink)
135                | (DirectionSpec::SinkToSource, Direction::SinkToSource)
136        )
137    }
138
139    #[must_use]
140    pub fn as_str(self) -> &'static str {
141        match self {
142            DirectionSpec::SourceToSink => "source-to-sink",
143            DirectionSpec::SinkToSource => "sink-to-source",
144            DirectionSpec::Both => "both",
145        }
146    }
147}
148
149impl fmt::Display for DirectionSpec {
150    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
151        f.write_str(self.as_str())
152    }
153}
154
155#[derive(Debug, Clone, PartialEq, Eq)]
156pub struct ParseDirectionError(pub String);
157
158impl fmt::Display for ParseDirectionError {
159    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
160        write!(
161            f,
162            "unknown direction {:?}; expected one of: source-to-sink, sink-to-source, both",
163            self.0
164        )
165    }
166}
167
168impl std::error::Error for ParseDirectionError {}
169
170impl FromStr for DirectionSpec {
171    type Err = ParseDirectionError;
172
173    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
174        match normalize(s.trim()).as_str() {
175            "sourcetosink" | "srctosink" | "forward" | "fwd" | "source" | "src" | "out" => {
176                Ok(DirectionSpec::SourceToSink)
177            }
178            "sinktosource" | "sinktosrc" | "reverse" | "rev" | "sink" | "in" => {
179                Ok(DirectionSpec::SinkToSource)
180            }
181            "both" | "bidi" | "bidirectional" | "duplex" | "all" => Ok(DirectionSpec::Both),
182            _ => Err(ParseDirectionError(s.to_string())),
183        }
184    }
185}
186
187/// A declared pipeline entry: which plugin, on which path, with what config.
188///
189/// In TOML the plugin's own options are flattened alongside `name` and
190/// `direction`:
191///
192/// ```toml
193/// [[plugin]]
194/// name = "tee"
195/// direction = "both"
196/// file = "dump.log"
197/// format = "hex"
198/// ```
199#[derive(Debug, Clone, Default, Deserialize, Serialize)]
200pub struct PluginSpec {
201    #[serde(alias = "plugin", alias = "use")]
202    pub name: String,
203    #[serde(default)]
204    pub direction: DirectionSpec,
205    /// A name for this instance, used in logs and as the default label for
206    /// stages that print one. Without it a stage is called after its plugin,
207    /// with `#n` appended when the same plugin appears twice on one path.
208    #[serde(default, rename = "as")]
209    pub alias: Option<String>,
210    /// Override the plugin's default placement. `true` runs this stage on its
211    /// own task behind a bounded channel.
212    #[serde(default)]
213    pub detach: Option<bool>,
214    /// Plugin-defined options, opaque to the host.
215    #[serde(flatten, default)]
216    pub config: Map<String, Value>,
217}
218
219impl PluginSpec {
220    pub fn new(name: impl Into<String>, direction: DirectionSpec) -> Self {
221        Self {
222            name: name.into(),
223            direction,
224            alias: None,
225            detach: None,
226            config: Map::new(),
227        }
228    }
229
230    #[must_use]
231    pub fn named(mut self, alias: impl Into<String>) -> Self {
232        self.alias = Some(alias.into());
233        self
234    }
235
236    #[must_use]
237    pub fn detached(mut self, detach: bool) -> Self {
238        self.detach = Some(detach);
239        self
240    }
241
242    #[must_use]
243    pub fn with(mut self, key: impl Into<String>, value: impl Into<Value>) -> Self {
244        self.config.insert(key.into(), value.into());
245        self
246    }
247
248    pub fn set(&mut self, key: impl Into<String>, value: impl Into<Value>) -> &mut Self {
249        self.config.insert(key.into(), value.into());
250        self
251    }
252}
253
254impl fmt::Display for PluginSpec {
255    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
256        match &self.alias {
257            Some(alias) => write!(f, "{} ({}:{})", alias, self.name, self.direction),
258            None => write!(f, "{}:{}", self.name, self.direction),
259        }
260    }
261}
tocat_api/lib.rs

tocat_api/
lib.rs
