Struct S3Source 

pub struct S3Source { /* private fields */ }

Coordinated object-storage backfill source. See the crate docs for the delivery model and the split identity/drift contract.

Implementations

impl S3Source

pub fn new(config: S3SourceConfig, io: Handle) -> S3Source

A source over config, doing its network I/O on io — pass the pipeline’s I/O runtime handle (Pipeline::io_handle).

The handle must belong to a multi-thread runtime that outlives the pipeline (the pipeline’s I/O runtime satisfies both): the source, its lanes, and the coordinator briefly block pipeline threads on it, which a current-thread runtime cannot drive.

pub fn from_component_config( section: &ComponentConfig, io: Handle, ) -> Result<S3Source, ConfigError>

Build from the pipeline’s opaque source: { s3: ... } section.

pub fn with_framer<F>(self, factory: F) -> S3Source

where F: Fn() -> Box<dyn RecordFramer> + Send + Sync + 'static,

Set the record framer that cuts each object’s byte stream into records. spate-s3 is a transport and owns no framing, so this is required: supply the framer for the objects’ format — e.g. spate-json’s NdjsonFramer for NDJSON — before the pipeline opens the source.

factory builds a fresh RecordFramer per object (framers are per-object stateful and each lane frames its own split). A framed source always emits one record per payload, so its FramingContract is PerRecord and the paired deserializer decodes a single unit.

pub fn with_coordinator( self, coordinator: Box<dyn SplitCoordinator>, ) -> S3Source

Hand the source its coordinator — the multi-instance seam. Build any SplitCoordinator at assembly time (e.g. StoreCoordinator over the NATS store from spate-coordination, with your own tuning) and inject it here; run more replicas of the same pipeline against the same backend and they share the backfill. Must be called before the pipeline opens the source.

Without it the source runs solo over an in-process store: correct and self-terminating, but progress is ephemeral — a restart replays the whole prefix (a startup WARN says so).

Trait Implementations

impl Debug for S3Source

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Drop for S3Source

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.

impl Source for S3Source

type Lane = S3Lane

The lane type this source produces.

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 — the source emits whole payloads and the deserializer owns framing (Kafka, and any source that does not frame).

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<S3Lane>, 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).

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 flush_commits(&mut self) -> Result<(), SourceError>

Synchronously flush stored positions (shutdown, revocation).