Merge DroppableEncoder and Encoder into a single facility (#11446)

Our RRD encoding facilities are currently implemented by two distinct
types: `DroppableEncoder` & `Encoder`.

This makes understanding how to use the API unnecessarily painful. Turns
out the only reason things are done this way is because of an
implementation detail related to the `Drop` implementation.
Instead of passing off that complexity to end-users of the API, we now
contain it to the implementation instead.

In addition to that, the `AlreadyFinished` logic was broken, so fix too
while we're at it.

---

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

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::DroppableEncoder;
+    use re_log_encoding::encoder::Encoder;
     use re_log_types::{LogMsg, SetStoreInfo, StoreId, StoreInfo, StoreKind, StoreSource};
 
     use super::*;
@@ -358,7 +358,7 @@ mod tests {
             .open(rrd_file_path.to_str().unwrap())
             .unwrap();
 
-        let mut encoder = DroppableEncoder::new(
+        let mut encoder = Encoder::new(
             re_build_info::CrateVersion::LOCAL,
             re_log_encoding::EncodingOptions::PROTOBUF_UNCOMPRESSED,
             rrd_file,

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

@@ -485,7 +485,7 @@ mod tests {
     use super::*;
     use crate::Compression;
     use crate::codec::arrow::encode_arrow;
-    use crate::encoder::DroppableEncoder;
+    use crate::encoder::Encoder;
 
     pub fn fake_log_messages() -> Vec<LogMsg> {
         let store_id = StoreId::random(StoreKind::Blueprint, "test_app");
@@ -676,7 +676,7 @@ mod tests {
         for options in options {
             let mut file = vec![];
 
-            let mut encoder = DroppableEncoder::new(rrd_version, options, &mut file).unwrap();
+            let mut encoder = Encoder::new(rrd_version, options, &mut file).unwrap();
             for message in messages.clone() {
                 encoder
                     .append_proto(message)
@@ -735,7 +735,7 @@ mod tests {
         for options in options {
             let mut file = vec![];
 
-            let mut encoder = DroppableEncoder::new(rrd_version, options, &mut file).unwrap();
+            let mut encoder = Encoder::new(rrd_version, options, &mut file).unwrap();
             for message in out_of_order_messages.clone() {
                 encoder
                     .append_proto(message)
@@ -824,23 +824,28 @@ mod tests {
             let messages = fake_log_messages();
 
             // (2 encoders as each encoder writes a file header)
-            let writer = std::io::Cursor::new(&mut data);
-            let mut encoder1 =
-                crate::encoder::Encoder::new(CrateVersion::LOCAL, options, writer).unwrap();
-            for message in &messages {
-                encoder1.append(message).unwrap();
+            {
+                let writer = std::io::Cursor::new(&mut data);
+                let mut encoder1 =
+                    crate::encoder::Encoder::new(CrateVersion::LOCAL, options, writer).unwrap();
+                for message in &messages {
+                    encoder1.append(message).unwrap();
+                }
+                encoder1.finish().unwrap();
             }
-            encoder1.finish().unwrap();
 
             let written = data.len() as u64;
-            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();
-            for message in &messages {
-                encoder2.append(message).unwrap();
+
+            {
+                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();
+                for message in &messages {
+                    encoder2.append(message).unwrap();
+                }
+                encoder2.finish().unwrap();
             }
-            encoder2.finish().unwrap();
 
             let decoder =
                 Decoder::new_concatenated(std::io::BufReader::new(data.as_slice())).unwrap();

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

@@ -330,15 +330,14 @@ 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 buffer = Vec::new();
-        let mut encoder = Encoder::new(CrateVersion::LOCAL, options, &mut buffer).unwrap();
+        let mut encoder = Encoder::new(CrateVersion::LOCAL, options, vec![]).unwrap();
         for message in &messages {
             encoder.append(message).unwrap();
         }
 
         encoder.finish().unwrap();
 
-        (messages, buffer)
+        (messages, encoder.into_inner().unwrap())
     }
 
     macro_rules! assert_message_ok {