Struct PipelineConfig 

#[non_exhaustive]
pub struct PipelineConfig {
    pub pipeline: PipelineSection,
    pub admin: AdminSection,
    pub backpressure: BackpressureSection,
    pub checkpoint: CheckpointSection,
    pub metrics: MetricsSection,
    pub source: ComponentConfig,
    pub deserializer: Option<ComponentConfig>,
    pub sink: Option<ComponentConfig>,
    pub sinks: Option<BTreeMap<String, ComponentConfig>>,
}

Root of a pipeline’s configuration file.

One process runs one pipeline; one file configures one process.

Construct with PipelineConfig::new, or PipelineConfig::new_multi_sink for a sinks: map, and set the optional fields. The struct is #[non_exhaustive] so new sections can be added without breaking callers.

Fields (Non-exhaustive)

Non-exhaustive structs could have additional fields added in future. Therefore, non-exhaustive structs cannot be constructed in external crates using the traditional Struct { .. } syntax; cannot be matched against without a wildcard ..; and struct update syntax will not work.

pipeline: PipelineSection

Identity and thread budget.

admin: AdminSection

The admin server carrying /metrics, /healthz and /readyz.

backpressure: BackpressureSection

In-flight budget and pause/resume hysteresis.

checkpoint: CheckpointSection

Watermark commit policy.

metrics: MetricsSection

Exporter selection and observability knobs.

source: ComponentConfig

The source component (opaque body).

deserializer: Option<ComponentConfig>

Optional deserializer component (opaque body). Sources that emit ready-made records need none.

sink: Option<ComponentConfig>

Single sink component (opaque body), sugar for the common one-sink case, addressed as "default". Mutually exclusive with sinks. Resolve via sink_config.

sinks: Option<BTreeMap<String, ComponentConfig>>

Named sinks for a multi-sink split: a name -> component map, each an ordinary single-key component (clickhouse: {...}). Mutually exclusive with sink. Resolve via sink_config.

Implementations

impl PipelineConfig

pub fn new( pipeline: PipelineSection, source: ComponentConfig, sink: ComponentConfig, ) -> PipelineConfig

A config for one source and one sink, the sink: form. Every optional section starts at its YAML default, and the sink is addressed as "default" by sink_config.

use spate_core::config::{ComponentConfig, PipelineConfig, PipelineSection, YamlValue};
use std::time::Duration;

let mut pipeline = PipelineSection::new("orders");
pipeline.io_threads = 4;

let mut cfg = PipelineConfig::new(
    pipeline,
    ComponentConfig::new("memory", YamlValue::Mapping(Default::default())),
    ComponentConfig::new("memory", YamlValue::Mapping(Default::default())),
);
cfg.checkpoint.interval = Duration::from_secs(10);
cfg.validate()?;

assert_eq!(cfg.sink_config("default")?.type_tag(), "memory");

pub fn new_multi_sink( pipeline: PipelineSection, source: ComponentConfig, sinks: BTreeMap<String, ComponentConfig>, ) -> PipelineConfig

A config for one source and a map of named sinks, the sinks: form. Every optional section starts at its YAML default, and each name addresses its sink through sink_config.

The map is not checked here. validate rejects an empty map, an empty name, and the reserved name "sink".

use spate_core::config::{ComponentConfig, PipelineConfig, PipelineSection, YamlValue};
use std::collections::BTreeMap;

let body = || YamlValue::Mapping(Default::default());
let sinks = BTreeMap::from([
    ("eu".to_owned(), ComponentConfig::new("memory", body())),
    ("us".to_owned(), ComponentConfig::new("memory", body())),
]);

let cfg = PipelineConfig::new_multi_sink(
    PipelineSection::new("orders"),
    ComponentConfig::new("memory", body()),
    sinks,
);
cfg.validate()?;

assert_eq!(cfg.sink_names(), ["eu", "us"]);

pub fn with_deserializer(self, deserializer: ComponentConfig) -> PipelineConfig

The same config with a deserializer component, tagged with its section the way the constructors tag the source and the sink.

use spate_core::config::{ComponentConfig, PipelineConfig, PipelineSection, YamlValue};

let body = || YamlValue::Mapping(Default::default());
let cfg = PipelineConfig::new(
    PipelineSection::new("orders"),
    ComponentConfig::new("memory", body()),
    ComponentConfig::new("memory", body()),
)
.with_deserializer(ComponentConfig::new("json", body()));

assert_eq!(
    cfg.deserializer.as_ref().map(ComponentConfig::type_tag),
    Some("json")
);

pub fn from_str(text: &str) -> Result<Self, ConfigError>

Load from YAML text: interpolate ${VAR} forms against the process environment, parse, and validate.

pub fn from_path(path: &Path) -> Result<Self, ConfigError>

Load from a YAML file (read, interpolate, parse, validate).

pub fn validate(&self) -> Result<(), ConfigError>

Cross-field validation, run automatically by the loaders. Public so programmatically built configs (tests, spate-test) get the same checks.

pub fn sink_config(&self, name: &str) -> Result<&ComponentConfig, ConfigError>

The component config for the sink named name. The single-sink sink: form is addressed as "default". A connector factory calls this once per sink to build it.

Errors

ConfigError::Validation if no sink is configured under name.

pub fn sink_names(&self) -> Vec<String>

The configured sink names, sorted. A single-sink config reports ["default"].

Trait Implementations

impl Debug for PipelineConfig

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<'de> Deserialize<'de> for PipelineConfig

fn deserialize<__D>(__deserializer: __D) -> Result<Self, __D::Error>

where __D: Deserializer<'de>,

Deserialize this value from the given Serde deserializer.

impl PartialEq for PipelineConfig

fn eq(&self, other: &PipelineConfig) -> bool

Equality operator ==.

fn ne(&self, other: &Rhs) -> bool

Inequality operator !=.

impl StructuralPartialEq for PipelineConfig