Struct Engine 

pub struct Engine { /* private fields */ }

The main Varpulis engine

Implementations

impl Engine

pub async fn process(&mut self, event: Event) -> Result<(), EngineError>

Process an incoming event (async-runtime only).

pub async fn process_shared( &mut self, event: SharedEvent, ) -> Result<(), EngineError>

Process a pre-wrapped SharedEvent (async-runtime only).

pub async fn process_batch( &mut self, events: Vec<Event>, ) -> Result<(), EngineError>

Process a batch of events for improved throughput (async-runtime only).

pub fn process_batch_sync( &mut self, events: Vec<Event>, ) -> Result<(), EngineError>

Synchronous batch processing for maximum throughput. Use this when no .to() sink operations are in the pipeline (e.g., benchmark mode). Avoids async runtime overhead completely.

pub async fn process_batch_shared( &mut self, events: Vec<SharedEvent>, ) -> Result<(), EngineError>

Process a batch of pre-wrapped SharedEvents (async-runtime only).

pub async fn flush_expired_sessions(&mut self) -> Result<(), EngineError>

Flush all expired session windows (async-runtime only).

impl Engine

pub fn builder() -> EngineBuilder

Create an EngineBuilder for fluent engine construction (async-runtime only).

pub fn new(output_tx: Sender<Event>) -> Self

Create engine with an owned output channel (async-runtime only).

pub fn new_benchmark() -> Self

Create engine without output channel (for benchmarking - skips Event cloning overhead). Available in both async and WASM builds.

pub fn new_with_optional_output(output_tx: Option<Sender<Event>>) -> Self

Create engine with optional output channel (async-runtime only, legacy API).

pub fn new_shared(output_tx: Sender<SharedEvent>) -> Self

Create engine with zero-copy SharedEvent output channel (async-runtime only).

pub fn set_credentials_store(&mut self, store: Arc<CredentialsStore>)

Set the connector credentials store for profile resolution (async-runtime only).

pub fn new_sync() -> Self

Constructor for sync-only / WASM builds (no tokio, no output channel). Outputs are collected internally and returned via process_batch_sync_collect().

pub fn process_batch_sync_collect( &mut self, events: Vec<Event>, ) -> Result<Vec<Event>, EngineError>

Process events synchronously and return collected outputs directly. Used by WASM builds and tests where mpsc channels are not needed.

pub fn set_context_name(&mut self, name: &str)

Set the context name for this engine instance.

pub fn set_topic_prefix(&mut self, prefix: &str)

Set a topic prefix for multi-tenant isolation (call before load()).

When set, all Kafka and MQTT topic names will be prefixed with {prefix}. to enforce per-tenant topic isolation.

pub fn set_trace_enabled(&mut self, enabled: bool)

Enable or disable pipeline trace collection.

When enabled, the engine records which streams each event is routed to, which operators pass or block the event, SASE pattern state, and emitted output events. Use drain_trace to retrieve entries.

pub fn is_trace_enabled(&self) -> bool

Whether pipeline trace collection is currently enabled.

pub fn drain_trace(&mut self) -> Vec<TraceEntry>

Drain all collected trace entries since the last drain.

pub fn set_dlq_path(&mut self, path: PathBuf)

Set a custom DLQ file path (call before load()).

pub const fn set_dlq_config(&mut self, config: DlqConfig)

Set custom DLQ configuration (call before load()).

pub const fn dlq(&self) -> Option<&Arc<DeadLetterQueue>>

Access the shared DLQ instance (created during load()).

pub fn get_pattern(&self, name: &str) -> Option<&NamedPattern>

Get a named pattern by name

pub const fn patterns(&self) -> &FxHashMap<String, NamedPattern>

Get all registered patterns

pub fn get_config(&self, name: &str) -> Option<&EngineConfig>

Get a configuration block by name

pub fn get_connector(&self, name: &str) -> Option<&ConnectorConfig>

Get a declared connector by name (async-runtime only).

pub const fn connector_configs(&self) -> &FxHashMap<String, ConnectorConfig>

Get all declared connector configs (async-runtime only).

pub fn source_bindings(&self) -> &[SourceBinding]

Get source connector bindings from .from() declarations

pub fn get_variable(&self, name: &str) -> Option<&Value>

Get a variable value by name

pub fn set_variable( &mut self, name: &str, value: Value, ) -> Result<(), EngineError>

Set a variable value (must be mutable or new)

pub const fn variables(&self) -> &FxHashMap<String, Value>

Get all variables (for debugging/testing)

pub const fn context_map(&self) -> &ContextMap

Get the context map (for orchestrator setup, async-runtime only).

pub fn has_contexts(&self) -> bool

Check if the loaded program declares any contexts (async-runtime only).

pub fn with_metrics(self, metrics: Metrics) -> Self

Enable Prometheus metrics (async-runtime only).

pub fn add_filter<F>( &mut self, stream_name: &str, filter: F, ) -> Result<(), EngineError>

where F: Fn(&Event) -> bool + Send + Sync + 'static,

Add a programmatic filter to a stream using a closure

pub fn load_with_source( &mut self, source: &str, program: &Program, ) -> Result<(), EngineError>

Load a program into the engine with semantic validation.

pub fn load(&mut self, program: &Program) -> Result<(), EngineError>

Load a program into the engine (no semantic validation).

pub async fn connect_sinks(&self) -> Result<(), EngineError>

Connect all sinks that require explicit connection (async-runtime only).

pub fn inject_sink(&mut self, key: &str, sink: Arc<dyn Sink>)

Inject a pre-built sink into the engine’s registry (async-runtime only).

pub fn has_sink(&self, key: &str) -> bool

Check whether a given key has a registered sink (async-runtime only).

pub fn sink_keys_for_connector(&self, connector_name: &str) -> Vec<String>

Return all sink keys that belong to a given connector name (async-runtime only).

pub fn has_sink_operations(&self) -> bool

Check if any registered stream uses .to() or .enrich() operations.

pub const fn event_counters(&self) -> (u64, u64)

Returns (events_in, events_out) counters for this engine.

pub fn is_stateless(&self) -> bool

Check if all streams are stateless (safe for round-robin distribution).

pub fn partition_key(&self) -> Option<String>

Return the partition key used by partitioned operations, if any.

pub fn has_session_windows(&self) -> bool

Check if any registered stream has session windows.

pub fn min_session_gap(&self) -> Option<Duration>

Return the smallest session gap across all streams (used as sweep interval).

pub fn metrics(&self) -> EngineMetrics

Get metrics

pub fn stream_names(&self) -> Vec<&str>

Get the names of all loaded streams.

pub fn physical_plan_summary(&self) -> Option<String>

Get the physical plan summary (available after load()).

pub fn explain(&self, program: &Program) -> Result<String, EngineError>

pub fn topology(&self) -> Topology

Build a topology snapshot of the currently loaded streams and routes.

pub fn get_function(&self, name: &str) -> Option<&UserFunction>

Get a user-defined function by name

pub fn function_names(&self) -> Vec<&str>

Get all registered function names

pub fn get_timers(&self) -> Vec<(u64, Option<u64>, String)>

Get all timer configurations for spawning timer tasks

pub fn register_scalar_udf(&mut self, udf: Arc<dyn ScalarUDF>)

Register a native scalar UDF.

pub fn register_aggregate_udf(&mut self, udf: Arc<dyn AggregateUDF>)

Register a native aggregate UDF.

pub const fn udf_registry(&self) -> &UdfRegistry

Get a reference to the UDF registry.

pub fn reload(&mut self, program: &Program) -> Result<ReloadReport, EngineError>

Reload program without losing state where possible (async-runtime only).

pub fn create_checkpoint(&self) -> EngineCheckpoint

Create a checkpoint of the engine state (windows, SASE engines, joins, variables).

pub fn restore_checkpoint( &mut self, cp: &EngineCheckpoint, ) -> Result<(), StoreError>

Restore engine state from a checkpoint.

pub fn enable_checkpointing( &mut self, store: Arc<dyn StateStore>, config: CheckpointConfig, ) -> Result<(), StoreError>

Enable auto-checkpointing with the given store and config.

pub fn checkpoint_tick(&mut self) -> Result<(), StoreError>

Check if a checkpoint is due and create one if needed.

pub fn force_checkpoint(&mut self) -> Result<(), StoreError>

Force an immediate checkpoint regardless of the interval.

pub const fn has_checkpointing(&self) -> bool

Returns true if auto-checkpointing is enabled.

pub fn enable_watermark_tracking(&mut self)

Enable per-source watermark tracking for this engine.

pub fn register_watermark_source(&mut self, source: &str, max_ooo: Duration)

Register a source for watermark tracking with its max out-of-orderness.

pub async fn advance_external_watermark( &mut self, source_context: &str, watermark_ms: i64, ) -> Result<(), EngineError>

Advance the watermark from an external source (async-runtime only).

Trait Implementations

impl Debug for Engine

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

Formats the value using the given formatter.