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::{Chain, Emitted, Pipeline, Registry, Segment},
58    plugin::{
59        BuildCtx, Ctx, EffectSink, Emission, Emit, Execution, ExternalStage, LogLevel,
60        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/// [`DirectionSpec::Both`] instantiates the plugin twice, once per direction,
106/// rather than sharing one instance between them.
107#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash, Serialize)]
108#[serde(rename_all = "kebab-case")]
109pub enum DirectionSpec {
110    SourceToSink,
111    SinkToSource,
112    #[default]
113    Both,
114}
115
116/// Deserialized through [`FromStr`] so that a config file accepts exactly what
117/// the command line does.
118impl<'de> Deserialize<'de> for DirectionSpec {
119    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> std::result::Result<Self, D::Error> {
120        let raw = String::deserialize(deserializer)?;
121        raw.parse().map_err(serde::de::Error::custom)
122    }
123}
124
125impl DirectionSpec {
126    #[must_use]
127    pub fn contains(self, direction: Direction) -> bool {
128        matches!(
129            (self, direction),
130            (DirectionSpec::Both, _)
131                | (DirectionSpec::SourceToSink, Direction::SourceToSink)
132                | (DirectionSpec::SinkToSource, Direction::SinkToSource)
133        )
134    }
135
136    #[must_use]
137    pub fn as_str(self) -> &'static str {
138        match self {
139            DirectionSpec::SourceToSink => "source-to-sink",
140            DirectionSpec::SinkToSource => "sink-to-source",
141            DirectionSpec::Both => "both",
142        }
143    }
144}
145
146impl fmt::Display for DirectionSpec {
147    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
148        f.write_str(self.as_str())
149    }
150}
151
152#[derive(Debug, Clone, PartialEq, Eq)]
153pub struct ParseDirectionError(pub String);
154
155impl fmt::Display for ParseDirectionError {
156    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
157        write!(
158            f,
159            "unknown direction {:?}; expected one of: source-to-sink, sink-to-source, both",
160            self.0
161        )
162    }
163}
164
165impl std::error::Error for ParseDirectionError {}
166
167impl FromStr for DirectionSpec {
168    type Err = ParseDirectionError;
169
170    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
171        match normalize(s.trim()).as_str() {
172            "sourcetosink" | "srctosink" | "forward" | "fwd" | "source" | "src" | "out" => {
173                Ok(DirectionSpec::SourceToSink)
174            }
175            "sinktosource" | "sinktosrc" | "reverse" | "rev" | "sink" | "in" => {
176                Ok(DirectionSpec::SinkToSource)
177            }
178            "both" | "bidi" | "bidirectional" | "duplex" | "all" => Ok(DirectionSpec::Both),
179            _ => Err(ParseDirectionError(s.to_string())),
180        }
181    }
182}
183
184/// A declared pipeline entry: which plugin, on which path, with what config.
185///
186/// In TOML the plugin's own options are flattened alongside `name` and
187/// `direction`:
188///
189/// ```toml
190/// [[plugin]]
191/// name = "tee"
192/// direction = "both"
193/// file = "dump.log"
194/// format = "hex"
195/// ```
196#[derive(Debug, Clone, Default, Deserialize, Serialize)]
197pub struct PluginSpec {
198    #[serde(alias = "plugin", alias = "use")]
199    pub name: String,
200    #[serde(default)]
201    pub direction: DirectionSpec,
202    /// A name for this instance, used in logs and as the default label for
203    /// stages that print one. Without it a stage is called after its plugin,
204    /// with `#n` appended when the same plugin appears twice on one path.
205    #[serde(default, rename = "as")]
206    pub alias: Option<String>,
207    /// Override the plugin's default placement. `true` runs this stage on its
208    /// own task behind a bounded channel.
209    #[serde(default)]
210    pub detach: Option<bool>,
211    /// Plugin-defined options, opaque to the host.
212    #[serde(flatten, default)]
213    pub config: Map<String, Value>,
214}
215
216impl PluginSpec {
217    pub fn new(name: impl Into<String>, direction: DirectionSpec) -> Self {
218        Self {
219            name: name.into(),
220            direction,
221            alias: None,
222            detach: None,
223            config: Map::new(),
224        }
225    }
226
227    #[must_use]
228    pub fn named(mut self, alias: impl Into<String>) -> Self {
229        self.alias = Some(alias.into());
230        self
231    }
232
233    #[must_use]
234    pub fn detached(mut self, detach: bool) -> Self {
235        self.detach = Some(detach);
236        self
237    }
238
239    #[must_use]
240    pub fn with(mut self, key: impl Into<String>, value: impl Into<Value>) -> Self {
241        self.config.insert(key.into(), value.into());
242        self
243    }
244
245    pub fn set(&mut self, key: impl Into<String>, value: impl Into<Value>) -> &mut Self {
246        self.config.insert(key.into(), value.into());
247        self
248    }
249}
250
251impl fmt::Display for PluginSpec {
252    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
253        match &self.alias {
254            Some(alias) => write!(f, "{} ({}:{})", alias, self.name, self.direction),
255            None => write!(f, "{}:{}", self.name, self.direction),
256        }
257    }
258}
tocat_api/lib.rs

tocat_api/
lib.rs
