utility: add support for encoding string slice (#4570)

* add support for encoding string slice

* add helpeful conversion from recordkey

* clippy fix

Author: sehz

crates/fluvio-protocol/src/core/encoder.rs

@@ -463,6 +463,42 @@ where
     }
 }
 
+impl Encoder for &str {
+    fn write_size(&self, _version: Version) -> usize {
+        2 + self.len()
+    }
+
+    fn encode<T>(&self, dest: &mut T, _version: Version) -> Result<(), Error>
+    where
+        T: BufMut,
+    {
+        if dest.remaining_mut() < 2 + self.len() {
+            return Err(Error::new(
+                ErrorKind::UnexpectedEof,
+                "not enough capacity for string",
+            ));
+        }
+
+        dest.put_u16(self.len() as u16);
+
+        let mut writer = dest.writer();
+        let bytes_written = writer.write(<str>::as_bytes(self))?;
+
+        if bytes_written != self.len() {
+            return Err(Error::new(
+                ErrorKind::UnexpectedEof,
+                format!(
+                    "out of {} bytes, {} not written",
+                    self.len(),
+                    self.len() - bytes_written
+                ),
+            ));
+        }
+
+        Ok(())
+    }
+}
+
 #[cfg(test)]
 mod test {
 

crates/fluvio-protocol/src/record/data.rs

@@ -80,6 +80,21 @@ impl RecordKey {
     }
 }
 
+impl From<RecordData> for RecordKey {
+    fn from(k: RecordData) -> Self {
+        Self(RecordKeyInner::Key(k))
+    }
+}
+
+impl From<RecordKey> for Option<RecordData> {
+    fn from(k: RecordKey) -> Self {
+        match k.0 {
+            RecordKeyInner::Key(data) => Some(data),
+            RecordKeyInner::Null => None,
+        }
+    }
+}
+
 #[derive(Hash)]
 enum RecordKeyInner {
     Null,
@@ -808,4 +823,15 @@ mod test {
         };
         assert_eq!(record.timestamp(), 1_000_000_800);
     }
+
+    #[test]
+    fn test_key_conversion() {
+        let null_key = RecordKey::NULL;
+        let data: Option<RecordData> = null_key.into();
+        assert_eq!(data, None);
+        let data = RecordData::from("test");
+        let key = RecordKey::from(data.clone());
+        let data2: Option<RecordData> = key.into();
+        assert_eq!(data2, Some(data));
+    }
 }

crates/fluvio-protocol/tests/str.rs

@@ -0,0 +1,16 @@
+use std::io::Cursor;
+
+use fluvio_protocol::{Decoder, Encoder};
+
+#[test]
+fn test_encode_str() {
+    let value = "hello";
+    assert_eq!(value.write_size(0), 7); // 7 bytes for the string "hello"
+    let mut dest = Vec::new();
+    value.encode(&mut dest, 0).expect("Failed to encode string");
+
+    // decode the string back to its original form
+    let mut buf = Cursor::new(dest);
+    let decoded = String::decode_from(&mut buf, 0).expect("Failed to decode string");
+    assert_eq!(decoded, "hello");
+}