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

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

Process a pre-wrapped SharedEvent (zero-copy path for context pipelines)

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

Process a batch of events for improved throughput. More efficient than calling process() repeatedly because:

• Pre-allocates SharedEvents in bulk
• Collects output events and sends in batches
• Amortizes async overhead

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 (zero-copy path for context pipelines)

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

Flush all expired session windows and process the resulting events through the remaining pipeline stages (aggregate, having, select, emit, etc.).

impl Engine

pub fn builder() -> EngineBuilder

Create an EngineBuilder for fluent engine construction.

Examples

use varpulis_runtime::Engine;
use tokio::sync::mpsc;

let (tx, _rx) = mpsc::channel(100);
let mut engine = Engine::builder()
    .output(tx)
    .build();

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

pub fn new_benchmark() -> Self

Create engine without output channel (for benchmarking - skips Event cloning overhead)

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

Create engine with optional output channel (legacy API, requires cloning)

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

Create engine with zero-copy SharedEvent output channel (PERF: avoids cloning)

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

Set the connector credentials store for profile resolution.

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_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

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

Get all declared connector configs (for building a ManagedConnectorRegistry).

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)

pub fn has_contexts(&self) -> bool

Check if the loaded program declares any contexts

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

Enable Prometheus metrics

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.

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

Inject a pre-built sink into the engine’s registry.

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

Check whether a given key has a registered sink.

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

Return all sink keys that belong to a given connector name.

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.

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 (e.g., upstream context).

Trait Implementations

impl Debug for Engine

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

Formats the value using the given formatter.