Struct WebsocketSource 

pub struct WebsocketSource { /* private fields */ }

Available on crate features source-rest and source-websocket only.

A WebSocket streaming source.

Implementations

impl WebsocketSource

pub fn new( config: WebsocketSourceConfig, ) -> Result<WebsocketSource, FaucetError>

Create a new WebSocket source. Validates the config; the connection is established lazily inside the stream loop (so reconnect can re-establish it mid-run).

pub fn with_auth_provider( self, provider: Arc<dyn AuthProvider>, ) -> WebsocketSource

Attach a shared AuthProvider. When set, the provider’s credential is resolved on every (re)connect — so a refreshed token is used automatically after reconnect. Takes precedence over inline auth.

Trait Implementations

impl Source for WebsocketSource

fn fetch_with_context<'life0, 'life1, 'async_trait>( &'life0 self, context: &'life1 HashMap<String, Value>, ) -> Pin<Box<dyn Future<Output = Result<Vec<Value>, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, WebsocketSource: 'async_trait,

Drain the entire run window into memory. This buffers every record the run produces (bounded only by max_messages / idle_timeout); prefer Source::stream_pages for large or long-running feeds so memory stays bounded at batch_size.

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

where 'life0: 'async_trait, 'life1: 'async_trait, WebsocketSource: 'async_trait,

Preflight probe that does not open the WebSocket stream.

The default Source::check would call stream_pages, which connects, sends subscribe frames, and then blocks waiting for inbound frames until max_messages / idle_timeout — useless as a fast preflight. Instead we only verify TCP reachability of the configured endpoint: parse the ws:///wss:// URL, resolve host + port (default 80 for ws, 443 for wss), open a tokio::net::TcpStream (no WS upgrade handshake, no auth, no frames), and immediately drop it.

fn stream_pages<'a>( &'a self, context: &'a HashMap<String, Value>, _batch_size: usize, ) -> Pin<Box<dyn Stream<Item = Result<StreamPage, FaucetError>> + Send + 'a>>

Stream records page-by-page so the pipeline can write to the sink as pages arrive instead of buffering the full result set.

fn config_schema(&self) -> Value

Return a JSON Schema describing the configuration this source accepts.

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

Stable identifier used as the connector label on metrics and the connector attribute on spans. Defaults to the final segment of std::any::type_name::<Self>(), e.g. "RestSource". Built-in connectors override with a short, friendly snake_case name (e.g. "rest"). Must return a non-empty string; observability decorators fall back to "unknown" in release builds if it is empty (and debug_assert! in debug builds).

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

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

Convenience: fetch with no parent context.

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

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

Incremental fetch with parent context support.

fn fetch_all_incremental<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<(Vec<Value>, Option<Value>), FaucetError>> + Send + 'async_trait>>

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

Convenience: incremental fetch with no parent context.

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

Stable key under which this source’s incremental-replication bookmark should be persisted in a StateStore.

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

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

Apply a bookmark loaded from a StateStore as this run’s starting point.