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,
impl<C> BaseProducer<C>where
C: ProducerContext,
Sourcepub fn poll<T: Into<Timeout>>(&self, timeout: T) -> i32
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.
Sourcepub fn send<'a, K, P>(
&self,
record: BaseRecord<'a, K, P, C::DeliveryOpaque>,
) -> Result<(), (KafkaError, BaseRecord<'a, K, P, C::DeliveryOpaque>)>
pub fn send<'a, K, P>( &self, record: BaseRecord<'a, K, P, C::DeliveryOpaque>, ) -> Result<(), (KafkaError, BaseRecord<'a, K, P, C::DeliveryOpaque>)>
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,
impl<C> Clone for BaseProducer<C>where
C: ProducerContext,
Source§fn clone(&self) -> BaseProducer<C>
fn clone(&self) -> BaseProducer<C>
1.0.0 · Source§fn clone_from(&mut self, source: &Self)
fn clone_from(&mut self, source: &Self)
source
. Read moreSource§impl FromClientConfig for BaseProducer<DefaultProducerContext>
impl FromClientConfig for BaseProducer<DefaultProducerContext>
Source§fn from_config(
config: &ClientConfig,
) -> KafkaResult<BaseProducer<DefaultProducerContext>>
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,
impl<C> FromClientConfigAndContext<C> for BaseProducer<C>where
C: ProducerContext,
Source§fn from_config_and_context(
config: &ClientConfig,
context: C,
) -> KafkaResult<BaseProducer<C>>
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,
impl<C> Producer<C> for BaseProducer<C>where
C: ProducerContext,
Source§fn flush<T: Into<Timeout>>(&self, timeout: T) -> KafkaResult<()>
fn flush<T: Into<Timeout>>(&self, timeout: T) -> KafkaResult<()>
Source§fn in_flight_count(&self) -> i32
fn in_flight_count(&self) -> i32
Source§fn init_transactions<T: Into<Timeout>>(&self, timeout: T) -> KafkaResult<()>
fn init_transactions<T: Into<Timeout>>(&self, timeout: T) -> KafkaResult<()>
Source§fn begin_transaction(&self) -> KafkaResult<()>
fn begin_transaction(&self) -> KafkaResult<()>
Source§fn send_offsets_to_transaction<T: Into<Timeout>>(
&self,
offsets: &TopicPartitionList,
cgm: &ConsumerGroupMetadata,
timeout: T,
) -> KafkaResult<()>
fn send_offsets_to_transaction<T: Into<Timeout>>( &self, offsets: &TopicPartitionList, cgm: &ConsumerGroupMetadata, timeout: T, ) -> KafkaResult<()>
Source§fn commit_transaction<T: Into<Timeout>>(&self, timeout: T) -> KafkaResult<()>
fn commit_transaction<T: Into<Timeout>>(&self, timeout: T) -> KafkaResult<()>
Source§fn abort_transaction<T: Into<Timeout>>(&self, timeout: T) -> KafkaResult<()>
fn abort_transaction<T: Into<Timeout>>(&self, timeout: T) -> KafkaResult<()>
Source§fn context(&self) -> &Arc<C>
fn context(&self) -> &Arc<C>
ProducerContext
used to create this
producer.Auto Trait Implementations§
impl<C> Freeze for BaseProducer<C>
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> BorrowMut<T> for Twhere
T: ?Sized,
impl<T> BorrowMut<T> for Twhere
T: ?Sized,
Source§fn borrow_mut(&mut self) -> &mut T
fn borrow_mut(&mut self) -> &mut T
Source§impl<T> CloneToUninit for Twhere
T: Clone,
impl<T> CloneToUninit for Twhere
T: Clone,
Source§unsafe fn clone_to_uninit(&self, dst: *mut T)
unsafe fn clone_to_uninit(&self, dst: *mut T)
clone_to_uninit
)