Trait SourceDriver

Source

pub trait SourceDriver: Send + Sync {
Show 22 methods    // Required methods
    fn connector_name(&self) -> &'static str;
    fn read<'a>(
        &'a self,
        calls: &'a CallSlot,
        context: &'a HashMap<String, Value>,
        batch_size: usize,
    ) -> PageStream<'a>;

    // Provided methods
    fn config_schema(&self) -> Value { ... }
    fn dataset_uri(&self) -> String { ... }
    fn call_context(&self) -> CallContext { ... }
    fn state_key(&self) -> Option<String> { ... }
    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 { ... }
    fn deterministic_replay(&self) -> bool { ... }
    fn consumes_destructively(&self) -> bool { ... }
    fn set_cancel(&self, cancel: CancellationToken) -> bool { ... }
    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 { ... }
    fn capabilities(&self, derived: Capabilities) -> Capabilities { ... }
    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 { ... }
    fn set_run_clock(&self, now: DateTime<Utc>) { ... }
    fn set_call_context(&self, ctx: CallContext) { ... }
    fn set_roundtrip_recorder(&self, recorder: Arc<RoundtripRecorder>) { ... }
    fn native_output_formats(&self) -> &'static [NativeFormat] { ... }
    fn read_native<'a>(
        &'a self,
        calls: &'a CallSlot,
        context: &'a HashMap<String, Value>,
        format: NativeFormat,
        batch_size: usize,
    ) -> NativeStream<'a> { ... }
    fn supports_columnar(&self) -> bool { ... }
    fn read_columnar<'a>(
        &'a self,
        calls: &'a CallSlot,
        context: &'a HashMap<String, Value>,
        batch_size: usize,
    ) -> ColumnarStream<'a> { ... }
    fn supports_discover(&self) -> bool { ... }
    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 { ... }
}

Expand description

A dataset’s read primitives (RFC 0014). Wrap it in a DriverSource to get a Source.

read streams the dataset in pages, each with the bookmark to resume after it (when there is one); every backend request goes through the CallSlot it is handed.

use faucet_core::driver::{DriverSource, PageStream, SourceDriver};
use faucet_core::{CallSlot, Source, StreamPage, Value, async_trait, json};
use std::collections::HashMap;

struct Numbers;

#[async_trait]
impl SourceDriver for Numbers {
    fn connector_name(&self) -> &'static str {
        "numbers"
    }
    fn read<'a>(
        &'a self,
        _calls: &'a CallSlot,
        _context: &'a HashMap<String, Value>,
        _batch_size: usize,
    ) -> PageStream<'a> {
        Box::pin(faucet_core::async_stream::stream! {
            yield Ok(StreamPage { records: vec![json!(1), json!(2)], bookmark: Some(json!(2)) });
        })
    }
}

let source = DriverSource::new(Numbers);
let (rows, bookmark) = source.fetch_all_incremental().await.unwrap();
assert_eq!(rows.len(), 2);
assert_eq!(bookmark, Some(json!(2)));

Required Methods§

Source

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

The connector’s name, the connector metric label.

Source

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.

Provided Methods§

Source

fn config_schema(&self) -> Value

The JSON Schema of the connector’s config.

Source

fn dataset_uri(&self) -> String

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

Source

fn call_context(&self) -> CallContext

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

Source

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

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

Source

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.

Source

fn deterministic_replay(&self) -> bool

Whether a resumed read replays exactly what follows the bookmark.

Source

fn consumes_destructively(&self) -> bool

Whether reading removes what was read (a queue).

Source

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.

Source

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.

Source

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

Adjust the capabilities DriverSource derived. Default: unchanged.

Source

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.

Source