Struct Source 

pub struct Source<T: Record> { /* private fields */ }

A streaming data source.

Sources are the entry point for data into a streaming pipeline. They provide a type-safe interface for pushing records and control signals (watermarks).

Thread Safety

Sources can be cloned to create multiple producers. The first clone triggers an automatic upgrade from SPSC to MPSC mode.

Example

let (source, sink) = streaming::create::<MyEvent>(config);

// Single producer (SPSC mode)
source.push(event1)?;

// Clone for multiple producers (MPSC mode)
let source2 = source.clone();
std::thread::spawn(move || {
    source2.push(event2)?;
});

Implementations

impl<T: Record> Source<T>

pub fn push(&self, record: T) -> Result<(), StreamingError>

Pushes a record to the source, blocking if necessary.

Errors

Returns StreamingError::ChannelClosed if the sink has been dropped. Returns StreamingError::ChannelFull if backpressure strategy is Reject.

pub fn try_push(&self, record: T) -> Result<(), TryPushError<T>>

Tries to push a record without blocking.

Errors

Returns TryPushError containing the record if the push failed.

pub fn push_batch(&self, records: &[T]) -> usize

where T: Clone,

Pushes multiple records, returning the number successfully pushed.

Stops at the first failure. Requires Clone because records are cloned before being sent.

Performance Warning

This method clones each record. For zero-clone batch insertion, use push_batch_drain which takes ownership via an iterator.

pub fn push_batch_drain<I>(&self, records: I) -> usize

where I: IntoIterator<Item = T>,

Pushes records from an iterator, consuming them (zero-clone).

Returns the number of records successfully pushed. Stops at the first failure (channel full or closed).

Example

let events = vec![event1, event2, event3];
let pushed = source.push_batch_drain(events.into_iter());

pub fn push_arrow(&self, batch: RecordBatch) -> Result<(), StreamingError>

Pushes an Arrow RecordBatch directly.

This is more efficient than pushing individual records when you already have data in Arrow format.

Errors

Returns StreamingError::SchemaMismatch if the batch schema doesn’t match. Returns StreamingError::ChannelClosed if the sink has been dropped.

pub fn watermark(&self, timestamp: i64)

Emits a watermark timestamp.

Watermarks signal that no events with timestamps less than or equal to this value will arrive in the future. This enables window triggers and garbage collection.

Watermarks are monotonically increasing - if a lower timestamp is passed, it will be ignored.

pub fn current_watermark(&self) -> i64

Returns the current watermark value.

pub fn schema(&self) -> SchemaRef

Returns the schema for this source.

pub fn name(&self) -> Option<&str>

Returns the source name, if configured.

pub fn is_mpsc(&self) -> bool

Returns true if the source is in MPSC mode.

pub fn mode(&self) -> ChannelMode

Returns the channel mode.

pub fn is_closed(&self) -> bool

Returns true if the sink has been dropped.

pub fn pending(&self) -> usize

Returns the number of pending items in the buffer.

pub fn capacity(&self) -> usize

Returns the buffer capacity.

pub fn sequence(&self) -> u64

Returns the current sequence number (total successful pushes).

pub fn sequence_counter(&self) -> Arc<AtomicU64>

Returns the shared sequence counter for checkpoint registration.

pub fn watermark_atomic(&self) -> Arc<AtomicI64>

Returns the shared watermark atomic for checkpoint registration.

Trait Implementations

impl<T: Record> Clone for Source<T>

fn clone(&self) -> Self

Clones the source, triggering automatic SPSC → MPSC upgrade.

Performance Warning

This method allocates a new Arc<SourceInner>. The first clone also triggers an upgrade from SPSC to MPSC mode, which adds synchronization overhead to all subsequent push operations.

For maximum performance with a single producer, avoid cloning the source. Use clones only when you genuinely need multiple producer threads.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<T: Record + Debug> Debug for Source<T>

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

Formats the value using the given formatter.