Struct FutureProducer

pub struct FutureProducer<C = DefaultClientContext, R = DefaultRuntime, Part = NoCustomPartitioner>
where
    Part: Partitioner,
    C: ClientContext + 'static,
{ /* private fields */ }

A producer that returns a Future for every message being produced.

Since message production in rdkafka is asynchronous, the caller cannot immediately know if the delivery of the message was successful or not. The FutureProducer provides this information in a Future, which will be completed once the information becomes available.

This producer has an internal polling thread and as such it doesn’t need to be polled. It can be cheaply cloned to get a reference to the same underlying producer. The internal polling thread will be terminated when the FutureProducer goes out of scope.

Implementations

impl<C, R> FutureProducer<C, R>

where C: ClientContext + 'static, R: AsyncRuntime,

pub async fn send<K, P, T>( &self, record: FutureRecord<'_, K, P>, queue_timeout: T, ) -> OwnedDeliveryResult

where K: ToBytes + ?Sized, P: ToBytes + ?Sized, T: Into<Timeout>,

Sends a message to Kafka, returning the result of the send.

The queue_timeout parameter controls how long to retry for if the librdkafka producer queue is full. Set it to Timeout::Never to retry forever or Timeout::After(0) to never block. If the timeout is reached and the queue is still full, an RDKafkaErrorCode::QueueFull error will be reported in the OwnedDeliveryResult.

Keep in mind that queue_timeout only applies to the first phase of the send operation. Once the message is queued, the underlying librdkafka client has separate timeout parameters that apply, like delivery.timeout.ms.

See also the FutureProducer::send_result method, which will not retry the queue operation if the queue is full.

pub fn send_result<'a, K, P>( &self, record: FutureRecord<'a, K, P>, ) -> Result<DeliveryFuture, (KafkaError, FutureRecord<'a, K, P>)>

where K: ToBytes + ?Sized, P: ToBytes + ?Sized,

Like FutureProducer::send, but if enqueuing fails, an error will be returned immediately, alongside the FutureRecord provided.

pub fn poll<T: Into<Timeout> + Send>(&self, timeout: T)

Polls the internal producer.

This is not normally required since the FutureProducer has a thread dedicated to calling poll regularly.

Trait Implementations

impl<C, R> Clone for FutureProducer<C, R>

where C: ClientContext + 'static,

fn clone(&self) -> FutureProducer<C, R>

Returns a copy of the value.

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

Performs copy-assignment from source.

impl<R> FromClientConfig for FutureProducer<DefaultClientContext, R>

where R: AsyncRuntime,

fn from_config<'life0, 'async_trait>( config: &'life0 ClientConfig, ) -> Pin<Box<dyn Future<Output = KafkaResult<FutureProducer<DefaultClientContext, R>>> + Send + 'async_trait>>

where 'life0: 'async_trait,

Creates a client from a client configuration. The default client context will be used.

impl<C, R> FromClientConfigAndContext<C> for FutureProducer<C, R>

where C: ClientContext + 'static, R: AsyncRuntime,

fn from_config_and_context<'life0, 'async_trait>( config: &'life0 ClientConfig, context: C, ) -> Pin<Box<dyn Future<Output = KafkaResult<FutureProducer<C, R>>> + Send + 'async_trait>>

where 'life0: 'async_trait,

Creates a client from a client configuration and a client context.

impl<C, R, Part> Producer<FutureProducerContext<C>, Part> for FutureProducer<C, R, Part>

where C: ClientContext + 'static, R: AsyncRuntime, Part: Partitioner,

fn client(&self) -> &Client<FutureProducerContext<C>>

Returns the Client underlying this producer.

fn flush<'life0, 'async_trait, T>( &'life0 self, timeout: T, ) -> Pin<Box<dyn Future<Output = KafkaResult<()>> + Send + 'async_trait>>

where T: 'async_trait + Into<Timeout> + Send, Self: 'async_trait, 'life0: 'async_trait,

Flushes any pending messages.

fn purge(&self, flags: PurgeConfig)

Purge messages currently handled by the producer instance.

fn in_flight_count(&self) -> i32

Returns the number of messages that are either waiting to be sent or are sent but are waiting to be acknowledged.

fn init_transactions<'life0, 'async_trait, T>( &'life0 self, timeout: T, ) -> Pin<Box<dyn Future<Output = KafkaResult<()>> + Send + 'async_trait>>

where T: 'async_trait + Into<Timeout> + Send, Self: 'async_trait, 'life0: 'async_trait,

Enable sending transactions with this producer.

fn begin_transaction(&self) -> KafkaResult<()>

Begins a new transaction.

fn send_offsets_to_transaction<'life0, 'life1, 'life2, 'async_trait, T>( &'life0 self, offsets: &'life1 TopicPartitionList, cgm: &'life2 ConsumerGroupMetadata, timeout: T, ) -> Pin<Box<dyn Future<Output = KafkaResult<()>> + Send + 'async_trait>>

where T: 'async_trait + Into<Timeout> + Send, Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Associates an offset commit operation with this transaction.

fn commit_transaction<'life0, 'async_trait, T>( &'life0 self, timeout: T, ) -> Pin<Box<dyn Future<Output = KafkaResult<()>> + Send + 'async_trait>>

where T: 'async_trait + Into<Timeout> + Send, Self: 'async_trait, 'life0: 'async_trait,

Commits the current transaction.

fn abort_transaction<'life0, 'async_trait, T>( &'life0 self, timeout: T, ) -> Pin<Box<dyn Future<Output = KafkaResult<()>> + Send + 'async_trait>>

where T: 'async_trait + Into<Timeout> + Send, Self: 'async_trait, 'life0: 'async_trait,

Aborts the current transaction.

fn context(&self) -> &Arc<C>

Returns a reference to the ProducerContext used to create this producer.