rust: Channel and Schema structs have IDs

.github/workflows/ci.yml

@@ -493,7 +493,7 @@ jobs:
           components: "rustfmt, clippy"
       - run: rustup target add wasm32-unknown-unknown
       - run: cargo fmt --all -- --check
-      - run: cargo clippy -- --no-deps
+      - run: cargo clippy --all-targets -- --no-deps
       - run: cargo clippy --no-default-features -- --no-deps
       - run: cargo clippy --no-default-features --features lz4 -- --no-deps
       - run: cargo clippy --no-default-features --features zstd -- --no-deps

rust/Cargo.toml

@@ -25,6 +25,7 @@ thiserror = "1.0"
 lz4 = { version = "1.27", optional = true }
 tokio = { version = "1", features = ["io-util"] , optional = true }
 static_assertions = "1.1.0"
+bimap = "0.6.3"
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 zstd = { version = "0.11", features = ["wasm"], optional = true }

rust/benches/reader.rs

@@ -17,12 +17,14 @@ fn create_test_mcap(n: usize, compression: Option<mcap::Compression>) -> Vec<u8>
         const MESSAGE_DATA: &[u8] = &[42; 10];
 
         let schema = Arc::new(Schema {
+            id: 1,
             name: "TestSchema".to_string(),
             encoding: "raw".to_string(),
             data: Cow::Borrowed(b"{}"),
         });
 
         let channel = Arc::new(Channel {
+            id: 0,
             topic: "test_topic".to_string(),
             message_encoding: "raw".to_string(),
             metadata: Default::default(),
@@ -35,7 +37,7 @@ fn create_test_mcap(n: usize, compression: Option<mcap::Compression>) -> Vec<u8>
                 sequence: i as u32,
                 log_time: i as u64,
                 publish_time: i as u64,
-                data: Cow::Borrowed(&MESSAGE_DATA),
+                data: Cow::Borrowed(MESSAGE_DATA),
             };
             writer.write(&message).unwrap();
         }

rust/examples/common/conformance_writer_spec.rs

@@ -20,28 +20,26 @@ pub struct Record {
 }
 
 impl Record {
-    pub fn get_field(self: &Self, name: &str) -> &Value {
-        return &self
+    pub fn get_field(&self, name: &str) -> &Value {
+        &self
             .fields
             .iter()
             .find(|f| f.0 == name)
             .unwrap_or_else(|| panic!("Invalid: {}", name))
-            .1;
+            .1
     }
 
-    pub fn get_field_data(self: &Self, name: &str) -> Vec<u8> {
-        let data: Vec<u8> = self
-            .get_field(name)
+    pub fn get_field_data(&self, name: &str) -> Vec<u8> {
+        self.get_field(name)
             .as_array()
             .unwrap_or_else(|| panic!("Invalid: {}", name))
-            .into_iter()
+            .iter()
             .filter_map(|v| v.as_u64())
             .filter_map(|n| u8::try_from(n).ok())
-            .collect();
-        return data;
+            .collect()
     }
 
-    pub fn get_field_meta(self: &Self, name: &str) -> BTreeMap<String, String> {
+    pub fn get_field_meta(&self, name: &str) -> BTreeMap<String, String> {
         let data = self
             .get_field(name)
             .as_object()
@@ -50,38 +48,34 @@ impl Record {
         for (key, value) in data.iter() {
             result.insert(key.to_string(), value.as_str().unwrap().to_string());
         }
-        return result;
+        result
     }
 
-    pub fn get_field_str(self: &Self, name: &str) -> &str {
-        return self
-            .get_field(name)
+    pub fn get_field_str(&self, name: &str) -> &str {
+        self.get_field(name)
             .as_str()
-            .unwrap_or_else(|| panic!("Invalid: {}", name));
+            .unwrap_or_else(|| panic!("Invalid: {}", name))
     }
 
-    pub fn get_field_u16(self: &Self, name: &str) -> u16 {
-        return self
-            .get_field(name)
+    pub fn get_field_u16(&self, name: &str) -> u16 {
+        self.get_field(name)
             .as_str()
             .and_then(|s| s.parse::<u16>().ok())
-            .unwrap_or_else(|| panic!("Invalid: {}", name));
+            .unwrap_or_else(|| panic!("Invalid: {}", name))
     }
 
-    pub fn get_field_u32(self: &Self, name: &str) -> u32 {
-        return self
-            .get_field(name)
+    pub fn get_field_u32(&self, name: &str) -> u32 {
+        self.get_field(name)
             .as_str()
             .and_then(|s| s.parse::<u32>().ok())
-            .unwrap_or_else(|| panic!("Invalid: {}", name));
+            .unwrap_or_else(|| panic!("Invalid: {}", name))
     }
 
-    pub fn get_field_u64(self: &Self, name: &str) -> u64 {
-        return self
-            .get_field(name)
+    pub fn get_field_u64(&self, name: &str) -> u64 {
+        self.get_field(name)
             .as_str()
             .and_then(|s| s.parse::<u64>().ok())
-            .unwrap_or_else(|| panic!("Invalid: {}", name));
+            .unwrap_or_else(|| panic!("Invalid: {}", name))
     }
 }
 

rust/examples/conformance_writer.rs

@@ -1,20 +1,22 @@
-use std::{borrow::Cow, collections::HashMap, env, io::Write, sync::Arc};
+use std::{
+    borrow::Cow,
+    collections::{BTreeMap, HashMap},
+    env,
+};
 
 #[path = "common/conformance_writer_spec.rs"]
 mod conformance_writer_spec;
 
 fn write_file(spec: &conformance_writer_spec::WriterSpec) {
-    let mut tmp = tempfile::NamedTempFile::new().expect("Couldn't open file");
-    let tmp_path = tmp.path().to_owned();
-    let out_buffer = std::io::BufWriter::new(&mut tmp);
     let mut writer = mcap::WriteOptions::new()
         .compression(None)
         .profile("")
-        .create(out_buffer)
+        .disable_seeking(true)
+        .create(binrw::io::NoSeek::new(std::io::stdout()))
         .expect("Couldn't create writer");
 
-    let mut channels = HashMap::<u16, mcap::Channel>::new();
-    let mut schemas = HashMap::<u64, mcap::Schema>::new();
+    let mut channel_ids = HashMap::new();
+    let mut schema_ids = HashMap::new();
 
     for record in &spec.records {
         match record.record_type.as_str() {
@@ -35,29 +37,26 @@ fn write_file(spec: &conformance_writer_spec::WriterSpec) {
             }
             "Channel" => {
                 let id = record.get_field_u16("id");
-                let schema_id = record.get_field_u64("schema_id");
+                let schema_id = record.get_field_u16("schema_id");
+                let output_schema_id = match schema_id {
+                    0 => 0,
+                    input_schema_id => {
+                        *schema_ids.get(&input_schema_id).expect("unknown schema ID")
+                    }
+                };
                 let topic = record.get_field_str("topic");
                 let message_encoding = record.get_field_str("message_encoding");
-                let schema = schemas.get(&schema_id).expect("Missing schema");
-                let channel = mcap::Channel {
-                    schema: Some(Arc::new(schema.to_owned())),
-                    topic: topic.to_string(),
-                    message_encoding: message_encoding.to_string(),
-                    metadata: std::collections::BTreeMap::new(),
-                };
-                writer
-                    .add_channel(&channel)
+                let returned_id = writer
+                    .add_channel(output_schema_id, topic, message_encoding, &BTreeMap::new())
                     .expect("Couldn't write channel");
-                channels.insert(id, channel);
+                channel_ids.insert(id, returned_id);
             }
             "ChunkIndex" => {
                 // written automatically
             }
             "DataEnd" => {
                 let data_section_crc = record.get_field_u32("data_section_crc");
-                let _data_end = mcap::records::DataEnd {
-                    data_section_crc: data_section_crc,
-                };
+                let _data_end = mcap::records::DataEnd { data_section_crc };
                 // write data end
             }
             "Footer" => {
@@ -82,15 +81,23 @@ fn write_file(spec: &conformance_writer_spec::WriterSpec) {
             }
             "Message" => {
                 let channel_id = record.get_field_u16("channel_id");
-                let channel = channels.get(&channel_id).expect("Unknown channel");
-                let message = mcap::Message {
-                    channel: Arc::new(channel.to_owned()),
-                    data: Cow::from(record.get_field_data("data")),
-                    log_time: record.get_field_u64("log_time"),
-                    publish_time: record.get_field_u64("publish_time"),
-                    sequence: record.get_field_u32("sequence"),
-                };
-                writer.write(&message).expect("Write message failed");
+                let data = record.get_field_data("data");
+                let log_time = record.get_field_u64("log_time");
+                let publish_time = record.get_field_u64("publish_time");
+                let sequence = record.get_field_u32("sequence");
+                writer
+                    .write_to_known_channel(
+                        &mcap::records::MessageHeader {
+                            channel_id: *channel_ids
+                                .get(&channel_id)
+                                .expect("message on unexpected channel ID"),
+                            log_time,
+                            publish_time,
+                            sequence,
+                        },
+                        &data,
+                    )
+                    .expect("Write message failed");
             }
             "Metadata" => {
                 let name = record.get_field_str("name");
@@ -104,14 +111,12 @@ fn write_file(spec: &conformance_writer_spec::WriterSpec) {
             "Schema" => {
                 let name = record.get_field_str("name");
                 let encoding = record.get_field_str("encoding");
-                let id = record.get_field_u64("id");
-                let data: Vec<u8> = record.get_field_data(&"data");
-                let schema = mcap::Schema {
-                    name: name.to_owned(),
-                    encoding: encoding.to_owned(),
-                    data: Cow::from(data),
-                };
-                schemas.insert(id, schema);
+                let id = record.get_field_u16("id");
+                let data: Vec<u8> = record.get_field_data("data");
+                let returned_id = writer
+                    .add_schema(name, encoding, &data)
+                    .expect("cannot write schema");
+                schema_ids.insert(id, returned_id);
             }
             "Statistics" => {
                 // written automatically
@@ -125,11 +130,6 @@ fn write_file(spec: &conformance_writer_spec::WriterSpec) {
     }
 
     writer.finish().expect("Couldn't finish");
-
-    let contents = std::fs::read(tmp_path).expect("Couldn't read output");
-    std::io::stdout()
-        .write(&contents)
-        .expect("Couldn't write output");
 }
 
 pub fn main() {

rust/src/lib.rs

@@ -39,14 +39,7 @@
 //!
 //!     // Channels and schemas are automatically assigned ID as they're serialized,
 //!     // and automatically deduplicated with `Arc` when deserialized.
-//!     let my_channel = Channel {
-//!         topic: String::from("cool stuff"),
-//!         schema: None,
-//!         message_encoding: String::from("application/octet-stream"),
-//!         metadata: BTreeMap::default()
-//!     };
-//!
-//!     let channel_id = out.add_channel(&my_channel)?;
+//!     let channel_id = out.add_channel(0, "cool stuff", "application/octet-stream", &BTreeMap::new())?;
 //!
 //!     out.write_to_known_channel(
 //!         &MessageHeader {
@@ -169,6 +162,7 @@ pub enum Compression {
 /// or hold its own buffer if it was decompressed from a chunk.
 #[derive(Clone, PartialEq, Eq, Hash)]
 pub struct Schema<'a> {
+    pub id: u16,
     pub name: String,
     pub encoding: String,
     pub data: Cow<'a, [u8]>,
@@ -186,6 +180,7 @@ impl fmt::Debug for Schema<'_> {
 /// Describes a channel which [Message]s are published to in an MCAP file
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct Channel<'a> {
+    pub id: u16,
     pub topic: String,
     pub schema: Option<Arc<Schema<'a>>>,
 

rust/src/read.rs

@@ -299,6 +299,7 @@ impl<'a> ChannelAccumulator<'a> {
         }
 
         let schema = Arc::new(Schema {
+            id: header.id,
             name: header.name.clone(),
             encoding: header.encoding,
             data,
@@ -329,6 +330,7 @@ impl<'a> ChannelAccumulator<'a> {
         };
 
         let channel = Arc::new(Channel {
+            id: chan.id,
             topic: chan.topic.clone(),
             schema,
             message_encoding: chan.message_encoding,