Skip to main content

SourceDriver

surreal_sync_runtime::pipeline

Trait SourceDriver 

Source 
pub trait SourceDriver: Send {
    type Position: Clone + Send + Sync + 'static;

    // Required methods
    fn poll_work<'life0, 'async_trait>(
        &'life0 mut self,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<PositionedEvent<Self::Position>>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn advance_watermark<'life0, 'async_trait>(
        &'life0 mut self,
        position: Self::Position,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;

    // Provided methods
    fn is_finished(&self) -> bool { ... }
    fn between_events<'life0, 'async_trait>(
        &'life0 mut self,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<ControlSignal>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait { ... }
    fn on_schema_refresh<'life0, 'async_trait>(
        &'life0 mut self,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait { ... }
    fn on_adhoc_snapshot<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 mut self,
        _tables: &'life1 [String],
        _apply: &'life2 dyn AdhocApply,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait { ... }
    fn stop_reason(&self) -> Option<StopReason> { ... }
    fn checkpoint_policy(&self) -> CheckpointPolicy { ... }
    fn persist_checkpoint<'life0, 'async_trait>(
        &'life0 mut self,
        _position: Self::Position,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait { ... }
    fn read_progress_for_persist<'life0, 'async_trait>(
        &'life0 mut self,
    ) -> Pin<Box<dyn Future<Output = Result<Option<Self::Position>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait { ... }
    fn note_sunk_events(&mut self, _count: u64) { ... }
}
Expand description

Source-facing incremental driver: work items + optional schema / ad-hoc / cancel / checkpoint hooks.

Mirrors the spirit of [interleaved_snapshot::WatermarkSource]: run_source_runtime owns the loop; the driver supplies poll / advance_watermark and optional extension points.

§Defaults

All hooks default to no-op / empty so a minimal CDC source only implements poll_work, advance_watermark, and optionally is_finished.

Required Associated Types§

Source

type Position: Clone + Send + Sync + 'static

Checkpoint / resume position type.

Required Methods§

Source

fn poll_work<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<Vec<PositionedEvent<Self::Position>>>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

Next work items (row and/or relation changes). May be empty on idle.

Source

fn advance_watermark<'life0, 'async_trait>( &'life0 mut self, position: Self::Position, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

Mark position sink-safe. May be in-memory only, broker offset commit, or both — durable store writes belong in persist_checkpoint unless policy is CheckpointPolicy::AdvanceOnly.

Provided Methods§

Source

fn is_finished(&self) -> bool

Whether the driver will produce no more work (EOF).

Source

fn between_events<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<Vec<ControlSignal>>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

Polled between apply cycles; return control signals to handle before the next poll_work.

Source

fn on_schema_refresh<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

Handle ControlSignal::SchemaRefresh. Default: no-op.

Source

fn on_adhoc_snapshot<'life0, 'life1, 'life2, 'async_trait>( &'life0 mut self, _tables: &'life1 [String], _apply: &'life2 dyn AdhocApply, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Handle ControlSignal::AdHocSnapshot.

apply exposes the runtime’s sink + pipeline/transformer + ApplyOpts so drivers can run snapshot writes through the same transform/sink path as the incremental loop (no private durability bypass).

Default: no-op.

Source

fn stop_reason(&self) -> Option<StopReason>

If Some, the runtime stops after draining in-flight apply work.

Source

fn checkpoint_policy(&self) -> CheckpointPolicy

Checkpoint persistence policy. Default: CheckpointPolicy::PersistAfterAdvance.

Source

fn persist_checkpoint<'life0, 'async_trait>( &'life0 mut self, _position: Self::Position, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

Persist a sink-safe checkpoint (called after successful sink + advance_watermark according to CheckpointPolicy). Default: no-op.

Source

fn read_progress_for_persist<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<Option<Self::Position>>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

Optional sink-safe position to persist when CheckpointPolicy::IntervalWhenDrained finds the apply window empty but nothing was advanced since the last persist (e.g. filtered-only binlog traffic that advanced the read cursor with no work items).

The runtime only consults this while fully drained, so returning the current read position is safe. Default: None (no filtered-only persist).

Source

fn note_sunk_events(&mut self, _count: u64)

Notify the driver that count input (pre-transform) events are accounted for before advance_watermark.

count is the batch’s poll/input size, not the post-transform sink length — so filter under-count and fan-out over-count cannot stall or premature-advance Kafka pending-acks, wal2json emitted/sunk gates, or similar watermarks.

Called after a successful sink apply, and also under FailurePolicy::Skip for batches that are advanced past without writing (so drivers that gate slot/cursor advance on sunk counts do not stall). Default: no-op.

Dyn Compatibility§

This trait is dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors§