Merge pull request #605 from fede1024/davidblewett/struct-setters

Add missing setters for some structs

Author: davidblewett

src/message.rs

@@ -623,6 +623,60 @@ impl OwnedMessage {
     pub fn detach_headers(&mut self) -> Option<OwnedHeaders> {
         self.headers.take()
     }
+
+    /// Replaces the [`OwnedHeaders`] on this `OwnedMessage`.
+    pub fn replace_headers(mut self, headers: Option<OwnedHeaders>) -> Self {
+        if let Some(headers) = headers {
+            self.headers.replace(headers);
+        } else {
+            self.headers = None;
+        }
+        self
+    }
+
+    /// Sets the payload for this `OwnedMessage`.
+    pub fn set_payload(mut self, payload: Option<Vec<u8>>) -> Self {
+        if let Some(payload) = payload {
+            self.payload.replace(payload);
+        } else {
+            self.payload = None;
+        }
+        self
+    }
+
+    /// Sets the key for this `OwnedMessage`.
+    pub fn set_key(mut self, key: Option<Vec<u8>>) -> Self {
+        if let Some(key) = key {
+            self.key.replace(key);
+        } else {
+            self.key = None;
+        }
+        self
+    }
+
+    /// Sets the topic for this `OwnedMessage`.
+    pub fn set_topic(mut self, topic: String) -> Self {
+        self.topic = topic;
+        self
+    }
+
+    /// Sets the timestamp for this `OwnedMessage`.
+    pub fn set_timestamp(mut self, timestamp: Timestamp) -> Self {
+        self.timestamp = timestamp;
+        self
+    }
+
+    /// Sets the partition for this `OwnedMessage`.
+    pub fn set_partition(mut self, partition: i32) -> Self {
+        self.partition = partition;
+        self
+    }
+
+    /// Sets the offset for this `OwnedMessage`.
+    pub fn set_offset(mut self, offset: i64) -> Self {
+        self.offset = offset;
+        self
+    }
 }
 
 impl Message for OwnedMessage {

src/producer/base_producer.rs

@@ -196,6 +196,18 @@ impl<'a, K: ToBytes + ?Sized, P: ToBytes + ?Sized, D: IntoOpaque> BaseRecord<'a,
         self.headers = Some(headers);
         self
     }
+
+    /// Sets the destination topic of the record.
+    pub fn topic(mut self, topic: &'a str) -> BaseRecord<'a, K, P, D> {
+        self.topic = topic;
+        self
+    }
+
+    /// Sets the delivery opaque of the record.
+    pub fn delivery_opaque(mut self, delivery_opaque: D) -> BaseRecord<'a, K, P, D> {
+        self.delivery_opaque = delivery_opaque;
+        self
+    }
 }
 
 impl<'a, K: ToBytes + ?Sized, P: ToBytes + ?Sized> BaseRecord<'a, K, P, ()> {