Trait Source 

pub trait Source: Send {
    type Lane: SourceLane;

    // Required methods
    fn open(&mut self, ctx: SourceCtx) -> Result<(), SourceError>;
    fn poll_events(
        &mut self,
        timeout: Duration,
    ) -> Result<SourceEvent<Self::Lane>, SourceError>;
    fn commit(
        &mut self,
        watermarks: &[(PartitionId, i64)],
    ) -> Result<(), SourceError>;

    // Provided methods
    fn component_type(&self) -> &str { ... }
    fn framing_contract(&self) -> FramingContract { ... }
    fn configure_feed(
        &mut self,
        section: &FeedSection,
        feed: Arc<dyn Feed>,
        origin: FeedOrigin,
    ) -> Result<(), ConfigError> { ... }
    fn flush_commits(&mut self) -> Result<(), SourceError> { ... }
    fn pause(&mut self, lanes: &[LaneId]) -> Result<(), SourceError> { ... }
    fn resume(&mut self, lanes: &[LaneId]) -> Result<(), SourceError> { ... }
    fn renew_held(&mut self) { ... }
    fn reading_on_stop(&mut self) -> bool { ... }
    fn stop_reading(&mut self) -> Result<(), SourceError> { ... }
}

Control plane of a source. Driven by the runtime’s controller from a single thread; lanes run on pipeline threads.

Required Associated Types

type Lane: SourceLane

The lane type this source produces.

Required Methods

fn open(&mut self, ctx: SourceCtx) -> Result<(), SourceError>

Connect and prepare. Called once before any other method.

fn poll_events( &mut self, timeout: Duration, ) -> Result<SourceEvent<Self::Lane>, SourceError>

Service control-plane work (rebalance callbacks, statistics) and return the next event, waiting at most timeout. Must be called regularly regardless of backpressure state.

fn commit( &mut self, watermarks: &[(PartitionId, i64)], ) -> Result<(), SourceError>

Store per-partition committable positions (each is the offset one past the last acknowledged record). Positions are durable per the source’s own policy (e.g. interval auto-commit of stored offsets).

The last commit of a stop is this call too. It runs on the controller thread, so a source that commits to a remote store bounds the call itself.

Provided Methods

fn component_type(&self) -> &str

The component_type metric label for this source (e.g. "kafka"), mirroring SinkParts::with_component_type on the sink side. It is also the namespace of the source’s custom-metrics Meter: declaring "kafka" scopes the source’s own families under spate_kafka_source_*. The default "source" is a reserved root, so a source that does not override this gets no custom Meter (its framework stage metrics are unaffected).

fn framing_contract(&self) -> FramingContract

How the payloads this source emits are framed, so the framework can pair it with a deserializer without the two being coordinated by hand (see FramingContract). A source that splits its own bytes into one record per payload returns FramingContract::PerRecord; the default is FramingContract::WholePayload, where the source emits whole payloads and the deserializer owns framing (Kafka, and any source that does not frame).

fn configure_feed( &mut self, section: &FeedSection, feed: Arc<dyn Feed>, origin: FeedOrigin, ) -> Result<(), ConfigError>

Accept the pipeline’s feed: section and the feed it names. The runtime calls this once, before open and before the pipeline threads start, and only when the section is present.

A feed-reading source keeps feed to read its work units from, and reads the section’s common keys, such as units_held. The backend’s own keys were already applied when feed was built. origin says whether code passed feed in, and so may publish into it outside the pipeline. The default rejects the section, so a source that reads no feed fails at startup when one is set.

fn flush_commits(&mut self) -> Result<(), SourceError>

Synchronously flush stored positions (shutdown, revocation).

fn pause(&mut self, lanes: &[LaneId]) -> Result<(), SourceError>

Stop fetching for lanes (backpressure). Optional capability: sources that cannot pause rely on bounded-queue pushback alone.

fn resume(&mut self, lanes: &[LaneId]) -> Result<(), SourceError>

Resume fetching for lanes.

fn renew_held(&mut self)

Renew the feed units the source holds, and nothing else.

The controller calls this on each pass of a revocation’s wait for its drain. That wait lasts up to checkpoint.drain_timeout and makes no poll_events call, so a source that holds leased units renews them here. Once a stop is set, the wait also asks reading_on_stop on each pass. Never waits. Settles, acknowledges and emits nothing: what the renewal finds is reported by the next poll_events. The default does nothing.

fn reading_on_stop(&mut self) -> bool

Whether the source is still reading units it holds after a graceful stop began. The controller asks on each pass between the stop and the drain, and begins the drain on the first false or at STOP_RESERVE before the deadline. Never waits. The default returns false, so the drain begins at once.

fn stop_reading(&mut self) -> Result<(), SourceError>

End the stop’s reading: hand back every unit not read to its end. Called once, after the last reading_on_stop on a graceful stop, or at once when the run fails, and before the pipeline threads drain. Never waits. The default does nothing.

Errors

The runtime logs an error and goes on with the drain; units it did not hand back return through redelivery.

Dyn Compatibility

This trait is dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors