pub struct BaseProducer<C = DefaultProducerContext>
where C: ProducerContext + 'static,
{ /* private fields */ }
Expand description

Lowest level Kafka producer.

The BaseProducer needs to be polled at regular intervals in order to serve queued delivery report callbacks (for more information, refer to the module-level documentation). This producer can be cheaply cloned to create a new reference to the same underlying producer.

§Example usage

This code will send a message to Kafka. No custom ProducerContext is specified, so the DefaultProducerContext will be used. To see how to use a producer context, refer to the examples in the examples folder.

use rdkafka::config::ClientConfig;
use rdkafka::producer::{BaseProducer, BaseRecord, Producer};
use std::time::Duration;

let producer: BaseProducer = ClientConfig::new()
    .set("bootstrap.servers", "kafka:9092")
    .create()
    .expect("Producer creation error");

producer.send(
    BaseRecord::to("destination_topic")
        .payload("this is the payload")
        .key("and this is a key"),
).expect("Failed to enqueue");

// Poll at regular intervals to process all the asynchronous delivery events.
for _ in 0..10 {
    producer.poll(Duration::from_millis(100));
}

// And/or flush the producer before dropping it.
producer.flush(Duration::from_secs(1));

Implementations§

source§

impl<C> BaseProducer<C>
where C: ProducerContext,

source

pub fn poll<T: Into<Timeout>>(&self, timeout: T) -> i32

Polls the producer, returning the number of events served.

Regular calls to poll are required to process the events and execute the message delivery callbacks.

source

pub fn send<'a, K, P>( &self, record: BaseRecord<'a, K, P, C::DeliveryOpaque> ) -> Result<(), (KafkaError, BaseRecord<'a, K, P, C::DeliveryOpaque>)>
where K: ToBytes + ?Sized, P: ToBytes + ?Sized,

Sends a message to Kafka.

Message fields such as key, payload, partition, timestamp etc. are provided to this method via a BaseRecord. If the message is correctly enqueued in the producer’s memory buffer, the method will take ownership of the record and return immediately; in case of failure to enqueue, the original record is returned, alongside an error code. If the message fails to be produced after being enqueued in the buffer, the ProducerContext::delivery method will be called asynchronously, with the provided ProducerContext::DeliveryOpaque.

When no partition is specified the underlying Kafka library picks a partition based on a hash of the key. If no key is specified, a random partition will be used. To correctly handle errors, the delivery callback should be implemented.

Note that this method will never block.

Trait Implementations§

source§

impl<C> Clone for BaseProducer<C>
where C: ProducerContext,

source§

fn clone(&self) -> BaseProducer<C>

Returns a copy of the value. Read more
1.0.0 · source§

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

Performs copy-assignment from source. Read more
source§

impl FromClientConfig for BaseProducer<DefaultProducerContext>

source§

fn from_config( config: &ClientConfig ) -> KafkaResult<BaseProducer<DefaultProducerContext>>

Creates a new BaseProducer starting from a configuration.

source§

impl<C> FromClientConfigAndContext<C> for BaseProducer<C>
where C: ProducerContext,

source§

fn from_config_and_context( config: &ClientConfig, context: C ) -> KafkaResult<BaseProducer<C>>

Creates a new BaseProducer starting from a configuration and a context.

source§

impl<C> Producer<C> for BaseProducer<C>
where C: ProducerContext,

source§

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

Returns the Client underlying this producer.
source§

fn flush<T: Into<Timeout>>(&self, timeout: T) -> KafkaResult<()>

Flushes any pending messages. Read more
source§

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.
source§

fn init_transactions<T: Into<Timeout>>(&self, timeout: T) -> KafkaResult<()>

Enable sending transactions with this producer. Read more
source§

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

Begins a new transaction. Read more
source§

fn send_offsets_to_transaction<T: Into<Timeout>>( &self, offsets: &TopicPartitionList, cgm: &ConsumerGroupMetadata, timeout: T ) -> KafkaResult<()>

Associates an offset commit operation with this transaction. Read more
source§

fn commit_transaction<T: Into<Timeout>>(&self, timeout: T) -> KafkaResult<()>

Commits the current transaction. Read more
source§

fn abort_transaction<T: Into<Timeout>>(&self, timeout: T) -> KafkaResult<()>

Aborts the current transaction. Read more
source§

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

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

Auto Trait Implementations§

§

impl<C> RefUnwindSafe for BaseProducer<C>
where C: RefUnwindSafe,

§

impl<C> Send for BaseProducer<C>

§

impl<C> Sync for BaseProducer<C>

§

impl<C> Unpin for BaseProducer<C>

§

impl<C> UnwindSafe for BaseProducer<C>
where C: RefUnwindSafe,

Blanket Implementations§

source§

impl<T> Any for T
where T: 'static + ?Sized,

source§

fn type_id(&self) -> TypeId

Gets the TypeId of self. Read more
source§

impl<T> Borrow<T> for T
where T: ?Sized,

source§

fn borrow(&self) -> &T

Immutably borrows from an owned value. Read more
source§

impl<T> BorrowMut<T> for T
where T: ?Sized,

source§

fn borrow_mut(&mut self) -> &mut T

Mutably borrows from an owned value. Read more
source§

impl<T> From<T> for T

source§

fn from(t: T) -> T

Returns the argument unchanged.

source§

impl<T> Instrument for T

source§

fn instrument(self, span: Span) -> Instrumented<Self>

Instruments this type with the provided Span, returning an Instrumented wrapper. Read more
source§

fn in_current_span(self) -> Instrumented<Self>

Instruments this type with the current Span, returning an Instrumented wrapper. Read more
source§

impl<T, U> Into<U> for T
where U: From<T>,

source§

fn into(self) -> U

Calls U::from(self).

That is, this conversion is whatever the implementation of From<T> for U chooses to do.

source§

impl<T> ToOwned for T
where T: Clone,

§

type Owned = T

The resulting type after obtaining ownership.
source§

fn to_owned(&self) -> T

Creates owned data from borrowed data, usually by cloning. Read more
source§

fn clone_into(&self, target: &mut T)

Uses borrowed data to replace owned data, usually by cloning. Read more
source§

impl<T, U> TryFrom<U> for T
where U: Into<T>,

§

type Error = Infallible

The type returned in the event of a conversion error.
source§

fn try_from(value: U) -> Result<T, <T as TryFrom<U>>::Error>

Performs the conversion.
source§

impl<T, U> TryInto<U> for T
where U: TryFrom<T>,

§

type Error = <U as TryFrom<T>>::Error

The type returned in the event of a conversion error.
source§

fn try_into(self) -> Result<U, <U as TryFrom<T>>::Error>

Performs the conversion.
source§

impl<T> WithSubscriber for T

source§

fn with_subscriber<S>(self, subscriber: S) -> WithDispatch<Self>
where S: Into<Dispatch>,

Attaches the provided Subscriber to this type, returning a WithDispatch wrapper. Read more
source§

fn with_current_subscriber(self) -> WithDispatch<Self>

Attaches the current default Subscriber to this type, returning a WithDispatch wrapper. Read more