krabka_client_producer/
record.rs

1//! Public record types. `ProducerRecord` is what the caller sends,
2//! `RecordMetadata` is what it gets back, and `Header` holds the per-record key
3//! and value pairs.
4
5use std::{
6    future::Future,
7    pin::Pin,
8    task::{Context, Poll},
9};
10
11use bytes::Bytes;
12use tokio::sync::oneshot;
13
14use crate::error::ProducerError;
15
16#[derive(Debug, Clone, Default, PartialEq, Eq)]
17pub struct ProducerRecord {
18    pub topic: String,
19    /// If `Some(p)`, the producer bypasses the partitioner and uses partition
20    /// `p`. `send` fails a negative `p` with
21    /// [`ProducerError::InvalidPartition`].
22    pub partition: Option<i32>,
23    pub key: Option<Bytes>,
24    pub value: Option<Bytes>,
25    pub headers: Vec<Header>,
26    /// If `None`, the producer fills in the current wall-clock time at
27    /// accumulator append time. `send` fails a negative value with
28    /// [`ProducerError::InvalidTimestamp`].
29    pub timestamp_ms: Option<i64>,
30}
31
32impl ProducerRecord {
33    /// Check the record as Kafka's `ProducerRecord` constructor does: first
34    /// the timestamp, then the partition.
35    ///
36    /// # Errors
37    ///
38    /// Returns [`ProducerError::InvalidTimestamp`] for a negative timestamp,
39    /// and [`ProducerError::InvalidPartition`] for a negative partition.
40    pub(crate) fn validate(&self) -> Result<(), ProducerError> {
41        if let Some(timestamp) = self.timestamp_ms.filter(|timestamp| *timestamp < 0) {
42            return Err(ProducerError::InvalidTimestamp(timestamp));
43        }
44        if let Some(partition) = self.partition.filter(|partition| *partition < 0) {
45            return Err(ProducerError::InvalidPartition(partition));
46        }
47        Ok(())
48    }
49}
50
51#[derive(Debug, Clone, PartialEq, Eq)]
52pub struct Header {
53    pub key: String,
54    pub value: Option<Bytes>,
55}
56
57/// What the producer learned about a record that the broker acknowledged.
58///
59/// Kafka's `RecordMetadata` carries the same fields.
60#[derive(Debug, Clone, PartialEq, Eq)]
61pub struct RecordMetadata {
62    /// The topic of the record.
63    pub topic: String,
64    pub partition: i32,
65    /// The offset of the record in the partition. It is -1 when the offset is
66    /// not known, which is always the case with `acks=0`. Kafka's
67    /// `RecordMetadata` constructor keeps a base offset of -1 and does not add
68    /// the index of the record in its batch.
69    pub offset: i64,
70    /// The log append time from the broker when the topic uses
71    /// `message.timestamp.type=LogAppendTime`, and else the create time of the
72    /// record. Kafka's `FutureRecordMetadata.timestamp` makes the same choice.
73    pub timestamp_ms: i64,
74    /// The size of the key in bytes, or -1 for a null key.
75    pub serialized_key_size: i32,
76    /// The size of the value in bytes, or -1 for a null value.
77    pub serialized_value_size: i32,
78}
79
80/// The delivery result of a record queued by [`crate::Producer::enqueue`].
81///
82/// Await the handle to get metadata or a producer error. Dropping it leaves
83/// the queued record in the producer; it does not cancel delivery.
84#[derive(Debug)]
85#[must_use = "await the handle to observe delivery errors"]
86pub struct DeliveryHandle {
87    receiver: oneshot::Receiver<Result<RecordMetadata, ProducerError>>,
88}
89
90impl DeliveryHandle {
91    pub(crate) fn new(receiver: oneshot::Receiver<Result<RecordMetadata, ProducerError>>) -> Self {
92        Self { receiver }
93    }
94}
95
96impl Future for DeliveryHandle {
97    type Output = Result<RecordMetadata, ProducerError>;
98
99    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
100        Pin::new(&mut self.receiver)
101            .poll(cx)
102            .map(|result| result.unwrap_or(Err(ProducerError::Closed)))
103    }
104}
105
106#[cfg(test)]
107mod tests {
108
109    use super::*;
110
111    #[test]
112    fn producer_record_default_is_empty() {
113        let r = ProducerRecord::default();
114        assert2::assert!(
115            r == ProducerRecord {
116                topic: String::new(),
117                partition: None,
118                key: None,
119                value: None,
120                headers: vec![],
121                timestamp_ms: None,
122            }
123        );
124    }
125
126    #[tokio::test]
127    async fn delivery_handle_reports_metadata_errors_and_closed_sender() {
128        let metadata = RecordMetadata {
129            topic: "events".into(),
130            partition: 2,
131            offset: 42,
132            timestamp_ms: 7,
133            serialized_key_size: -1,
134            serialized_value_size: 5,
135        };
136        let (sender, receiver) = oneshot::channel();
137        let mut handle = DeliveryHandle::new(receiver);
138        assert2::assert!(futures::FutureExt::now_or_never(&mut handle).is_none());
139        sender.send(Ok(metadata.clone())).expect("handle is open");
140        assert2::assert!(handle.await.expect("delivery succeeds") == metadata);
141
142        for (result, expected) in [
143            (Some(Err(ProducerError::Server(42))), "broker error_code 42"),
144            (None, "producer closed"),
145        ] {
146            let (sender, receiver) = oneshot::channel();
147            let handle = DeliveryHandle::new(receiver);
148            if let Some(result) = result {
149                sender.send(result).expect("handle is open");
150            } else {
151                drop(sender);
152            }
153            assert2::assert!(handle.await.expect_err("delivery fails").to_string() == expected);
154        }
155    }
156}
krabka_client_producer/record.rs

krabka_client_producer/
record.rs
