Struct NatsSource 

pub struct NatsSource { /* private fields */ }

NATS source — core and JetStream modes.

Implementations

impl NatsSource

pub fn new(schema: SchemaRef, registry: Option<&Registry>) -> Self

Metrics register on registry if provided.

pub fn config(&self) -> Option<&NatsSourceConfig>

Available after SourceConnector::start.

pub fn metrics_handle(&self) -> &NatsSourceMetrics

Snapshot accessor for the prometheus-backed metrics struct.

Trait Implementations

impl SourceConnector for NatsSource

fn terminal_task_tracker(&self) -> Option<ConnectorTaskTracker>

Observe detached tasks whose lifetime may outlast this connector value.

fn contract( &self, _config: &ConnectorConfig, ) -> Result<SourceContract, ConnectorError>

Declare recovery and placement semantics for this exact configuration.

fn start<'life0, 'async_trait>( &'life0 mut self, request: SourceStart, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>

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

Opens the source and establishes its initial or resumed position as one indivisible lifecycle transition.

fn poll_batch<'life0, 'async_trait>( &'life0 mut self, max_records: usize, ) -> Pin<Box<dyn Future<Output = Result<Option<SourceBatch>, ConnectorError>> + Send + 'async_trait>>

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

Ok(None) = no data currently available; runtime retries after a delay. max_records is the normal batching target. A source may exceed it only when one upstream atomic replay unit cannot be split without making its checkpoint cursor invalid. Such sources must enforce independent hard record and byte limits and fail before retained data can grow unbounded.

fn schema(&self) -> SchemaRef

Arrow schema of records this source produces.

fn checkpoint(&self) -> SourceCheckpoint

Returned checkpoint must contain enough info to resume from the current position after a restart.

fn close<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>

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

Close the connection and release resources.

fn data_ready_notify(&self) -> Option<Arc<Notify>>

Returns a [Notify] handle that is signalled when new data is available.

fn cancellation_policy(&self) -> ConnectorCancellationPolicy

Deadline behavior required by the underlying client implementation.

fn discover_schema<'life0, 'life1, 'async_trait>( &'life0 mut self, _properties: &'life1 HashMap<String, String>, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>

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

Resolve the source schema from the WITH (...) properties before DDL reaches the planner. Implementations that perform network I/O must bound it with a timeout. Return Err(ConnectorError::…) on failure so the runtime can surface the cause to DDL — do not log and swallow.

fn try_checkpoint(&self) -> Result<Option<SourceCheckpoint>, ConnectorError>

Atomically attempt to capture a source cursor. Ok(None) means a transient control-plane publication is not reconciled yet; the runtime retains the barrier and retries without advancing it.

fn checkpoint_ready(&self) -> Result<bool, ConnectorError>

Whether this source’s data-plane cursor is reconciled with the current control-plane ownership publication. The runtime does not poll records or admit checkpoint barriers while this is false.

fn drive_control_plane(&mut self)

Start or advance connector-side control-plane work needed to become checkpoint_ready. Called even while data polling and barriers are fenced.

fn begin_drain( &mut self, _request: &SourceDrainRequest, _deadline: Instant, ) -> Result<(), ConnectorError>

Stop advancing external input for an exact cluster transition.

fn poll_drain_ready( &mut self, _round: AssignmentDrainId, ) -> Result<bool, ConnectorError>

Whether the exact provider FIFO boundary has been consumed.

fn finish_drain<'life0, 'async_trait>( &'life0 mut self, _resolution: SourceDrainResolution, _deadline: Instant, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>

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

Resolve an exact drain after target commit or abort.

fn set_vnode_assignment( &mut self, _source_identity: &str, _registry: Arc<VnodeRegistry>, _self_id: NodeId, ) -> Result<(), ConnectorError>

Install the cluster vnode assignment for a source that advertises SourceTopology::Splittable. The source identity is the stable, canonical catalog object name and must be part of any external split mapping ABI.

fn recovery_identity_options( &self, _config: &ConnectorConfig, ) -> Result<Option<BTreeMap<String, String>>, ConnectorError>

Return connector-owned semantic options for durable recovery identity.

fn notify_epoch_committed<'life0, 'life1, 'async_trait>( &'life0 mut self, _epoch: u64, _checkpoint: &'life1 SourceCheckpoint, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>

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

Acknowledge that epoch has been durably committed.