Encoder: simplified APIs and improved discoverability (#11450)

The Encoder was offloading so much complexity to end-users of the API by
forcing them to somehow discover, pick and figure out how to use a bunch
of complicated freestanding helpers such as `encode_ref_as_bytes_local`.

All these freestanding functions have now been replaced by a couple of
simpler, more generic, more discoverable helper methods.

---

This PR is part of an upcoming series of PRs to pay off organic growth
debt in our encoding/decoding stack.

* DNM: requires #11446

Author: teh-cmc

crates/store/re_data_loader/src/loader_rrd.rs

@@ -329,7 +329,7 @@ impl RetryableFileReader {
 #[cfg(test)]
 mod tests {
     use re_chunk::RowId;
-    use re_log_encoding::encoder::Encoder;
+    use re_log_encoding::Encoder;
     use re_log_types::{LogMsg, SetStoreInfo, StoreId, StoreInfo, StoreKind, StoreSource};
 
     use super::*;

crates/store/re_entity_db/examples/memory_usage.rs

@@ -70,7 +70,7 @@ fn log_messages() {
     fn encode_log_msg(log_msg: &LogMsg) -> Vec<u8> {
         let mut bytes = vec![];
         let encoding_options = re_log_encoding::EncodingOptions::PROTOBUF_COMPRESSED;
-        re_log_encoding::encoder::encode_ref(
+        re_log_encoding::Encoder::encode_into(
             re_build_info::CrateVersion::LOCAL,
             encoding_options,
             std::iter::once(log_msg).map(Ok),

crates/store/re_log_encoding/benches/msg_encode_benchmark.rs

@@ -38,7 +38,7 @@ fn encode_log_msgs(
     encoding_options: re_log_encoding::EncodingOptions,
 ) -> Vec<u8> {
     let mut bytes = vec![];
-    re_log_encoding::encoder::encode_ref(
+    re_log_encoding::Encoder::encode_into(
         re_build_info::CrateVersion::LOCAL,
         encoding_options,
         messages.iter().map(Ok),

crates/store/re_log_encoding/src/decoder/mod.rs

@@ -484,8 +484,8 @@ mod tests {
 
     use super::*;
     use crate::Compression;
+    use crate::Encoder;
     use crate::codec::arrow::encode_arrow;
-    use crate::encoder::Encoder;
 
     pub fn fake_log_messages() -> Vec<LogMsg> {
         let store_id = StoreId::random(StoreKind::Blueprint, "test_app");
@@ -643,7 +643,7 @@ mod tests {
 
         for options in options {
             let mut file = vec![];
-            crate::encoder::encode_ref(rrd_version, options, messages.iter().map(Ok), &mut file)
+            crate::Encoder::encode_into(rrd_version, options, messages.iter().map(Ok), &mut file)
                 .unwrap();
 
             let decoded_messages = Decoder::new(&mut file.as_slice())
@@ -787,7 +787,7 @@ mod tests {
 
         for options in options {
             let mut file = vec![];
-            crate::encoder::encode_ref(
+            crate::Encoder::encode_into(
                 rrd_version,
                 options,
                 out_of_order_messages.iter().map(Ok),
@@ -827,7 +827,7 @@ mod tests {
             {
                 let writer = std::io::Cursor::new(&mut data);
                 let mut encoder1 =
-                    crate::encoder::Encoder::new(CrateVersion::LOCAL, options, writer).unwrap();
+                    crate::Encoder::new(CrateVersion::LOCAL, options, writer).unwrap();
                 for message in &messages {
                     encoder1.append(message).unwrap();
                 }
@@ -840,7 +840,7 @@ mod tests {
                 let mut writer = std::io::Cursor::new(&mut data);
                 writer.set_position(written);
                 let mut encoder2 =
-                    crate::encoder::Encoder::new(CrateVersion::LOCAL, options, writer).unwrap();
+                    crate::Encoder::new(CrateVersion::LOCAL, options, writer).unwrap();
                 for message in &messages {
                     encoder2.append(message).unwrap();
                 }

crates/store/re_log_encoding/src/decoder/stream.rs

@@ -312,8 +312,8 @@ mod tests {
     use re_chunk::RowId;
     use re_log_types::{SetStoreInfo, StoreInfo};
 
+    use crate::Encoder;
     use crate::EncodingOptions;
-    use crate::encoder::Encoder;
 
     use super::*;
 
@@ -330,14 +330,16 @@ mod tests {
     fn test_data(options: EncodingOptions, n: usize) -> (Vec<LogMsg>, Vec<u8>) {
         let messages: Vec<_> = (0..n).map(|_| fake_log_msg()).collect();
 
-        let mut encoder = Encoder::new(CrateVersion::LOCAL, options, vec![]).unwrap();
-        for message in &messages {
-            encoder.append(message).unwrap();
-        }
-
-        encoder.finish().unwrap();
+        let mut data = Vec::new();
+        Encoder::encode_into(
+            CrateVersion::LOCAL,
+            options,
+            messages.clone().into_iter().map(Ok),
+            &mut data,
+        )
+        .unwrap();
 
-        (messages, encoder.into_inner().unwrap())
+        (messages, data)
     }
 
     macro_rules! assert_message_ok {

crates/store/re_log_encoding/src/decoder/streaming.rs

@@ -473,7 +473,7 @@ mod tests {
 
         for options in options {
             let mut data = vec![];
-            crate::encoder::encode_ref(rrd_version, options, messages.iter().map(Ok), &mut data)
+            crate::Encoder::encode_into(rrd_version, options, messages.iter().map(Ok), &mut data)
                 .unwrap();
 
             // We cut the input file by one byte to simulate a corrupted file and check that we don't end up in an infinite loop
@@ -517,7 +517,7 @@ mod tests {
 
         for options in options {
             let mut data = vec![];
-            crate::encoder::encode_ref(rrd_version, options, messages.iter().map(Ok), &mut data)
+            crate::Encoder::encode_into(rrd_version, options, messages.iter().map(Ok), &mut data)
                 .unwrap();
 
             let buf_reader = tokio::io::BufReader::new(std::io::Cursor::new(data));
@@ -557,7 +557,7 @@ mod tests {
 
         for options in options {
             let mut data = vec![];
-            crate::encoder::encode_ref(rrd_version, options, messages.iter().map(Ok), &mut data)
+            crate::Encoder::encode_into(rrd_version, options, messages.iter().map(Ok), &mut data)
                 .unwrap();
 
             let buf_reader = tokio::io::BufReader::new(std::io::Cursor::new(data.clone()));

crates/store/re_log_encoding/src/encoder.rs

@@ -1,5 +1,7 @@
 //! Encoding of [`LogMsg`]es as a binary stream, e.g. to store in an `.rrd` file, or send over network.
 
+use std::borrow::Borrow;
+
 use re_build_info::CrateVersion;
 use re_chunk::{ChunkError, ChunkResult};
 use re_log_types::LogMsg;
@@ -74,6 +76,33 @@ pub struct Encoder<W: std::io::Write> {
     is_finished: bool,
 }
 
+impl Encoder<Vec<u8>> {
+    pub fn local() -> Result<Self, EncodeError> {
+        Self::new(
+            CrateVersion::LOCAL,
+            EncodingOptions::PROTOBUF_COMPRESSED,
+            Vec::new(),
+        )
+    }
+
+    /// All-in-one helper to encode a stream of [`LogMsg`]s into an actual RRD stream.
+    ///
+    /// This always uses the local version and its default encoding options.
+    ///
+    /// Returns the encoded data in a newly allocated vector.
+    pub fn encode(
+        messages: impl IntoIterator<Item = ChunkResult<impl Borrow<LogMsg>>>,
+    ) -> Result<Vec<u8>, EncodeError> {
+        re_tracing::profile_function!();
+        let mut encoder = Self::local()?;
+        for message in messages {
+            encoder.append(message?.borrow())?;
+        }
+        encoder.finish()?;
+        encoder.into_inner()
+    }
+}
+
 impl<W: std::io::Write> Encoder<W> {
     pub fn new(
         version: CrateVersion,
@@ -191,6 +220,26 @@ impl<W: std::io::Write> Encoder<W> {
     }
 }
 
+impl<W: std::io::Write> Encoder<W> {
+    /// All-in-one helper to encode a stream of [`LogMsg`]s into an actual RRD stream.
+    ///
+    /// Returns the size in bytes of the encoded data.
+    pub fn encode_into(
+        version: CrateVersion,
+        options: EncodingOptions,
+        messages: impl IntoIterator<Item = ChunkResult<impl Borrow<LogMsg>>>,
+        write: &mut W,
+    ) -> Result<u64, EncodeError> {
+        re_tracing::profile_function!();
+        let mut encoder = Encoder::new(version, options, write)?;
+        let mut size_bytes = 0;
+        for message in messages {
+            size_bytes += encoder.append(message?.borrow())?;
+        }
+        Ok(size_bytes)
+    }
+}
+
 // TODO(cmc): It seems a bit suspicious to me that we send an EOS marker on drop, but don't flush.
 // But I don't want to change any flushing behavior at the moment, so I'll keep it that way for now.
 impl<W: std::io::Write> std::ops::Drop for Encoder<W> {
@@ -205,91 +254,3 @@ impl<W: std::io::Write> std::ops::Drop for Encoder<W> {
         }
     }
 }
-
-/// Returns the size in bytes of the encoded data.
-pub fn encode(
-    version: CrateVersion,
-    options: EncodingOptions,
-    messages: impl Iterator<Item = ChunkResult<LogMsg>>,
-    write: &mut impl std::io::Write,
-) -> Result<u64, EncodeError> {
-    re_tracing::profile_function!();
-    let mut encoder = Encoder::new(version, options, write)?;
-    let mut size_bytes = 0;
-    for message in messages {
-        size_bytes += encoder.append(&message?)?;
-    }
-    Ok(size_bytes)
-}
-
-/// Returns the size in bytes of the encoded data.
-pub fn encode_ref<'a>(
-    version: CrateVersion,
-    options: EncodingOptions,
-    messages: impl Iterator<Item = ChunkResult<&'a LogMsg>>,
-    write: &mut impl std::io::Write,
-) -> Result<u64, EncodeError> {
-    re_tracing::profile_function!();
-    let mut encoder = Encoder::new(version, options, write)?;
-    let mut size_bytes = 0;
-    for message in messages {
-        size_bytes += encoder.append(message?)?;
-    }
-    Ok(size_bytes)
-}
-
-pub fn encode_as_bytes(
-    version: CrateVersion,
-    options: EncodingOptions,
-    messages: impl Iterator<Item = ChunkResult<LogMsg>>,
-) -> Result<Vec<u8>, EncodeError> {
-    re_tracing::profile_function!();
-    let mut encoder = Encoder::new(version, options, vec![])?;
-    for message in messages {
-        encoder.append(&message?)?;
-    }
-    encoder.finish()?;
-    encoder.into_inner()
-}
-
-#[inline]
-pub fn local_encoder() -> Result<Encoder<Vec<u8>>, EncodeError> {
-    Encoder::new(
-        CrateVersion::LOCAL,
-        EncodingOptions::PROTOBUF_COMPRESSED,
-        Vec::new(),
-    )
-}
-
-#[inline]
-pub fn local_raw_encoder() -> Result<Encoder<Vec<u8>>, EncodeError> {
-    Encoder::new(
-        CrateVersion::LOCAL,
-        EncodingOptions::PROTOBUF_COMPRESSED,
-        Vec::new(),
-    )
-}
-
-#[inline]
-pub fn encode_as_bytes_local(
-    messages: impl Iterator<Item = ChunkResult<LogMsg>>,
-) -> Result<Vec<u8>, EncodeError> {
-    let mut encoder = local_raw_encoder()?;
-    for message in messages {
-        encoder.append(&message?)?;
-    }
-    encoder.finish()?;
-    encoder.into_inner()
-}
-
-#[inline]
-pub fn encode_ref_as_bytes_local<'a>(
-    messages: impl Iterator<Item = ChunkResult<&'a LogMsg>>,
-) -> Result<Vec<u8>, EncodeError> {
-    let mut encoder = local_raw_encoder()?;
-    for message in messages {
-        encoder.append(message?)?;
-    }
-    encoder.finish()?;
-    encoder.into_inner()
-}

crates/store/re_log_encoding/src/file_sink.rs

@@ -95,11 +95,8 @@ impl FileSink {
 
         let file = std::fs::File::create(&path)
             .map_err(|err| FileSinkError::CreateFile(path.clone(), err))?;
-        let encoder = crate::encoder::Encoder::new(
-            re_build_info::CrateVersion::LOCAL,
-            encoding_options,
-            file,
-        )?;
+        let encoder =
+            crate::Encoder::new(re_build_info::CrateVersion::LOCAL, encoding_options, file)?;
         let join_handle = spawn_and_stream(Some(&path), encoder, rx)?;
 
         Ok(Self {
@@ -117,7 +114,7 @@ impl FileSink {
 
         re_log::debug!("Writing to stdout…");
 
-        let encoder = crate::encoder::Encoder::new(
+        let encoder = crate::Encoder::new(
             re_build_info::CrateVersion::LOCAL,
             encoding_options,
             std::io::stdout(),
@@ -158,7 +155,7 @@ impl FileSink {
 /// Set `filepath` to `None` to stream to standard output.
 fn spawn_and_stream<W: std::io::Write + Send + 'static>(
     filepath: Option<&std::path::Path>,
-    mut encoder: crate::encoder::Encoder<W>,
+    mut encoder: crate::Encoder<W>,
     rx: Receiver<Option<Command>>,
 ) -> Result<std::thread::JoinHandle<()>, FileSinkError> {
     let (name, target) = if let Some(filepath) = filepath {

crates/store/re_log_encoding/src/lib.rs

@@ -4,7 +4,7 @@
 pub mod decoder;
 
 #[cfg(feature = "encoder")]
-pub mod encoder;
+mod encoder;
 
 mod app_id_injector;
 pub mod codec;
@@ -24,13 +24,16 @@ pub mod external {
 
 // ---------------------------------------------------------------------
 
-pub use app_id_injector::{
+pub use self::app_id_injector::{
     ApplicationIdInjector, CachingApplicationIdInjector, DummyApplicationIdInjector,
 };
 
+#[cfg(feature = "encoder")]
+pub use self::encoder::{EncodeError, Encoder};
+
 #[cfg(feature = "encoder")]
 #[cfg(not(target_arch = "wasm32"))]
-pub use file_sink::{FileFlushError, FileSink, FileSinkError};
+pub use self::file_sink::{FileFlushError, FileSink, FileSinkError};
 
 // ----------------------------------------------------------------------------
 

crates/store/re_log_encoding/tests/arrow_encode_roundtrip.rs

@@ -1,8 +1,7 @@
 use similar_asserts::assert_eq;
 
-use re_build_info::CrateVersion;
 use re_chunk::{Chunk, RowId, TimePoint, Timeline};
-use re_log_encoding::{EncodingOptions, decoder::decode_bytes, encoder::encode_as_bytes};
+use re_log_encoding::{Encoder, decoder::decode_bytes};
 use re_log_types::{LogMsg, StoreId};
 use re_types::archetypes::Points3D;
 
@@ -39,13 +38,7 @@ fn encode_roundtrip() {
     let store_id = StoreId::empty_recording();
     let messages = [LogMsg::ArrowMsg(store_id, arrow_msg)];
 
-    let option = EncodingOptions::PROTOBUF_COMPRESSED;
-    let crate_version = CrateVersion::LOCAL;
-    let encoded = encode_as_bytes(crate_version, option, messages.iter().cloned().map(Ok)).unwrap();
+    let encoded = Encoder::encode(messages.iter().cloned().map(Ok)).unwrap();
     let decoded = decode_bytes(&encoded).unwrap();
-    similar_asserts::assert_eq!(
-        decoded,
-        messages,
-        "Failed to roundtrip chunk with option {option:?}"
-    );
+    similar_asserts::assert_eq!(decoded, messages, "Failed to roundtrip chunk");
 }