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_coordination(
        &mut self,
        section: &CoordinationSection,
    ) -> Result<(), ConfigError> { ... }
    fn commit_final(
        &mut self,
        watermarks: &[(PartitionId, i64)],
    ) -> Result<Vec<PartitionId>, SourceError> { ... }
    fn flush_commits(&mut self) -> Result<(), SourceError> { ... }
    fn pause(&mut self, lanes: &[LaneId]) -> Result<(), SourceError> { ... }
    fn resume(&mut self, lanes: &[LaneId]) -> 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).

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_coordination( &mut self, section: &CoordinationSection, ) -> Result<(), ConfigError>

👎Deprecated since 0.3.0:

removed in v0.4.0, where a work feed replaces split coordination; see https://github.com/spate-etl/spate/issues/1088

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

A coordinated source validates the section here and builds its coordinator from it in open. The default rejects the section, so a source that does not coordinate fails at startup when one is set.

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

👎Deprecated since 0.3.0:

removed in v0.4.0, where a work feed replaces split coordination; see https://github.com/spate-etl/spate/issues/1088

The last commit of a stop, which a source committing to a remote store may bound as a whole.

Ok lists the partitions among watermarks that the source still holds and whose commit it could not confirm; the runtime counts every other position as committed, as commit does for a partition the source no longer holds. Err counts none of them as committed, and a fatal error fails the run. An unconfirmed commit may or may not have been stored, and positions past the stored one may replay.

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.

Dyn Compatibility

This trait is dyn compatible.

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

Implementors