Struct MessageSourceDriver 

pub struct MessageSourceDriver<C> { /* private fields */ }

A MessageConsumer as a SourceDriver.

Each page carries a bookmark, so the pipeline flushes the sink after writing it; the page’s messages are acknowledged when the next page is requested (or the stream is polled to its end) — that is, only once the pipeline has written and flushed them. A run that fails before then acknowledges nothing it did not write, and the broker redelivers it.

Implementations

impl<C: MessageConsumer> MessageSourceDriver<C>

pub fn new(consumer: C) -> Self

The source driver over consumer.

pub fn consumer(&self) -> &C

The consumer.

Trait Implementations

impl<C: MessageConsumer> SourceDriver for MessageSourceDriver<C>

fn connector_name(&self) -> &'static str

The connector’s name, the connector metric label.

fn config_schema(&self) -> Value

The JSON Schema of the connector’s config.

fn dataset_uri(&self) -> String

The dataset this source reads, for lineage (credential-free).

fn call_context(&self) -> CallContext

The connector’s own call defaults. The run’s context is layered over it.

fn consumes_destructively(&self) -> bool

Whether reading removes what was read (a queue).

fn read<'a>( &'a self, calls: &'a CallSlot, _context: &'a HashMap<String, Value>, batch_size: usize, ) -> PageStream<'a>

Stream the dataset in pages of about batch_size records (0: one page). context holds the parent record of a fan-out row.

fn fetch<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, calls: &'life1 CallSlot, _context: &'life2 HashMap<String, Value>, ) -> Pin<Box<dyn Future<Output = Option<Result<(Vec<Value>, Option<Value>), FaucetError>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

A buffered read with its own all-at-once path, returning the records and the bookmark to resume after them. None (the default): the buffered fetches drain read.

fn check<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, calls: &'life1 CallSlot, ctx: &'life2 CheckContext, ) -> Pin<Box<dyn Future<Output = Result<Option<CheckReport>, FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

A non-mutating preflight probe. None (the default) reads one page.

fn state_key(&self) -> Option<String>

The state key bookmarks are stored under, when the source resumes.

fn resume_from<'life0, 'async_trait>( &'life0 self, bookmark: Value, ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Resume after bookmark on the next read.

fn deterministic_replay(&self) -> bool

Whether a resumed read replays exactly what follows the bookmark.

fn set_cancel(&self, cancel: CancellationToken) -> bool

Take the run’s cancellation token. Return true when the source then yields what it buffered and ends, so the pipeline polls it to its end.

fn capabilities(&self, derived: Capabilities) -> Capabilities

Adjust the capabilities DriverSource derived. Default: unchanged.

fn set_run_clock(&self, now: DateTime<Utc>)

Receive the run clock (the instant ${now.*} renders from).

fn set_call_context(&self, ctx: CallContext)

The run’s call context, for a driver whose clients hold their own.

fn set_roundtrip_recorder(&self, recorder: Arc<RoundtripRecorder>)

The run’s round-trip recorder, for a driver whose clients hold their own.

fn native_output_formats(&self) -> &'static [NativeFormat]

Wire formats read_native can emit (#633).

fn read_native<'a>( &'a self, calls: &'a CallSlot, context: &'a HashMap<String, Value>, format: NativeFormat, batch_size: usize, ) -> NativeStream<'a>

Stream the dataset as raw bytes in a negotiated format.

fn supports_columnar(&self) -> bool

Available on crate feature arrow only.

Whether read_columnar is implemented.

fn read_columnar<'a>( &'a self, calls: &'a CallSlot, context: &'a HashMap<String, Value>, batch_size: usize, ) -> ColumnarStream<'a>

Available on crate feature arrow only.

Stream the dataset as Arrow batches.

fn supports_discover(&self) -> bool

Whether discover is implemented.

fn discover<'life0, 'life1, 'async_trait>( &'life0 self, calls: &'life1 CallSlot, ) -> Pin<Box<dyn Future<Output = Result<Vec<DatasetDescriptor>, FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Enumerate the datasets behind the connection (read-only, cheap).