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 resolve_schema<'life0, 'life1, 'async_trait>( &'life0 mut self, config: &'life1 ConnectorConfig, explicit: Option<SchemaRef>, ) -> Pin<Box<dyn Future<Output = Result<SchemaBinding, ConnectorError>> + Send + 'async_trait>>

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

Resolve an immutable reader contract before activation. Metadata adapters must override this hook to retain native identity/definitions and validate projections. The default preserves explicit columns and reuses pre-open discovery for fixed schemas.

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

Declare recovery and placement semantics for this exact configuration. This hook must be deterministic, configuration-only, and free of external I/O; admission and checkpoint identity inspect it before start().

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 supports_initialized_start(&self) -> bool

Whether atomic start validates and installs SourcePosition::Initialized without resolving its boundary again or acknowledging pre-boundary input. Discovery/validation hooks alone do not certify startup. The runtime rejects unrecognized implementations before I/O; this capability is neither topology authority nor actor readiness.

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 connector and format 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 resolve_initial_position<'life0, 'life1, 'async_trait>( &'life0 mut self, _config: &'life1 ConnectorConfig, ) -> Pin<Box<dyn Future<Output = Result<SourceCheckpoint, ConnectorError>> + Send + 'async_trait>>

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

Resolve a complete, unowned initial cursor without starting or consuming the source. The topology control path seals the first successful vector and reuses it on every retry. Implementations must bound metadata I/O and task ownership, avoid group joins, external acknowledgements and resource creation, and return the existing connector cursor encoding. The cursor is not a checkpoint of processed data or an intake permit. Ordinary startup and recovery keep their current policy until target installation explicitly consumes this cut.

fn validate_initial_position<'life0, 'life1, 'life2, 'async_trait>( &'life0 mut self, _config: &'life1 ConnectorConfig, _checkpoint: &'life2 SourceCheckpoint, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>

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

Read-only validation of a previously sealed global new-source cursor. Must preserve its numeric boundary and create no active reader, acknowledgement or sink effect. Connectors must reject a changed inventory or expired position rather than silently reset. This is preparation only; startup must validate again under final installation authority.

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_unavailable_policy(&self) -> SourceCheckpointUnavailablePolicy

Choose how the runtime progresses after Self::try_checkpoint returns Ok(None).

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 crate::connector::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 recency( &self, _config: &ConnectorConfig, ) -> Result<Option<Arc<dyn SourceRecency>>, ConnectorError>

How far the upstream of a source configured by config has been written, for reads that must include it. Configuration only, no external I/O. The default, None, means the connector cannot tell, and such reads are refused.

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.