producer

Implements RetryConfig and BaseProducer, the base class for all producer implementations.

class unistream.producer.RetryConfig(exp_backoff: list[int] = <factory>, attempts: int = 0, first_attempt_time: ~datetime.datetime | None = None, last_attempt_time: ~datetime.datetime | None = None, last_error: Exception | None = None)

The retry behavior configuration for BaseProducer.

Parameters:

• exp_backoff – the exponential backoff retry waiter settings. for example, if exp_backoff = [1, 2, 4], then wait 1 seconds before the second attempt, wait 2 seconds before the third attempt, wait 4 seconds before the fourth attempt, and then always wait 4 seconds for the next attempts. Default to [1, 2, 4, 8, 15, 30, 60].

• attempts – total attempts we have made.

• first_attempt_time – the time when we first attempt to send the data.

• last_attempt_time – the time when we last attempt to send the data.

• last_error – the last error we encountered.

shall_we_retry(now: datetime) → bool

Check whether we should retry sending the data to the sink.

class unistream.producer.BaseProducer(buffer: AbcBuffer = REQ, retry_config: RetryConfig = REQ)

Base class for producer implementations.

Provides buffer management and exponential-backoff retry logic. Subclasses only need to implement send().

Parameters:

• buffer – the AbcBuffer backend for batching records.

• retry_config – the RetryConfig for send retry behavior.

put(record: AbcRecord, skip_error: bool = True, verbose: bool = False)

Put the record to the buffer and smartly decide whether to send the records.