jetstreamer_firehose/
dataframe.rs

1use {
2    crate::{SharedError, node::Kind, node_reader::cid_from_cbor_link, utils::Buffer},
3    cid::Cid,
4    std::vec::Vec,
5};
6
7// type DataFrame struct {
8// 	Kind  int
9// 	Hash  **int
10// 	Index **int
11// 	Total **int
12// 	Data  []uint8
13// 	Next  **List__Link
14// }
15/// Representation of a `Kind::DataFrame` node from the firehose.
16#[derive(Clone, PartialEq, Eq, Hash, Debug)]
17pub struct DataFrame {
18    /// Kind discriminator copied from the CBOR payload.
19    pub kind: u64,
20    /// Optional rolling hash for the data chunk.
21    pub hash: Option<u64>,
22    /// Optional chunk index within the full payload.
23    pub index: Option<u64>,
24    /// Optional total number of chunks.
25    pub total: Option<u64>,
26    /// Raw bytes contained in this chunk.
27    pub data: Buffer,
28    /// Optional list of CIDs pointing to continuation chunks.
29    pub next: Option<Vec<Cid>>,
30}
31
32impl DataFrame {
33    /// Decodes a [`DataFrame`] from raw CBOR bytes.
34    pub fn from_bytes(data: Vec<u8>) -> Result<DataFrame, SharedError> {
35        let decoded_data: serde_cbor::Value = serde_cbor::from_slice(&data).unwrap();
36        let data_frame = DataFrame::from_cbor(decoded_data)?;
37        Ok(data_frame)
38    }
39
40    /// Decodes a [`DataFrame`] from a parsed CBOR [`serde_cbor::Value`].
41    pub fn from_cbor(val: serde_cbor::Value) -> Result<DataFrame, SharedError> {
42        let mut data_frame = DataFrame {
43            kind: 0,
44            hash: None,
45            index: None,
46            total: None,
47            data: Buffer::new(),
48            next: None,
49        };
50
51        if let serde_cbor::Value::Array(array) = val {
52            // println!("Kind: {:?}", array[0]);
53            if let Some(serde_cbor::Value::Integer(kind)) = array.first() {
54                // println!("Kind: {:?}", Kind::from_u64(kind as u64).unwrap().to_string());
55                data_frame.kind = *kind as u64;
56
57                if *kind as u64 != Kind::DataFrame as u64 {
58                    return Err(Box::new(std::io::Error::other(std::format!(
59                        "Wrong kind for DataFrame. Expected {:?}, got {:?}",
60                        Kind::DataFrame,
61                        kind
62                    ))) as SharedError);
63                }
64            }
65            if let Some(serde_cbor::Value::Integer(hash)) = array.get(1) {
66                data_frame.hash = Some(*hash as u64);
67            }
68            if let Some(serde_cbor::Value::Integer(index)) = array.get(2) {
69                data_frame.index = Some(*index as u64);
70            }
71            if let Some(serde_cbor::Value::Integer(total)) = array.get(3) {
72                data_frame.total = Some(*total as u64);
73            }
74            if let Some(serde_cbor::Value::Bytes(data)) = &array.get(4) {
75                data_frame.data = Buffer::from_vec(data.clone());
76            }
77
78            if array.len() > 5
79                && let Some(serde_cbor::Value::Array(next)) = &array.get(5)
80            {
81                if next.is_empty() {
82                    data_frame.next = None;
83                } else {
84                    let mut nexts = vec![];
85                    for cid in next {
86                        nexts.push(cid_from_cbor_link(cid)?);
87                    }
88                    data_frame.next = Some(nexts);
89                }
90            }
91        }
92        Ok(data_frame)
93    }
94
95    /// Renders the dataframe as a JSON value for debugging.
96    pub fn to_json(&self) -> serde_json::Value {
97        let mut next = vec![];
98        if let Some(nexts) = &self.next {
99            for cid in nexts {
100                next.push(serde_json::json!({
101                    "/": cid.to_string()
102                }));
103            }
104        }
105
106        let mut map = serde_json::Map::new();
107        map.insert("kind".to_string(), serde_json::Value::from(self.kind));
108        if self.hash.is_none() {
109            map.insert("hash".to_string(), serde_json::Value::Null);
110        } else {
111            let hash_as_string = self.hash.unwrap().to_string();
112            map.insert("hash".to_string(), serde_json::Value::from(hash_as_string));
113        }
114        if self.index.is_none() {
115            map.insert("index".to_string(), serde_json::Value::Null);
116        } else {
117            map.insert("index".to_string(), serde_json::Value::from(self.index));
118        }
119        if self.total.is_none() {
120            map.insert("total".to_string(), serde_json::Value::Null);
121        } else {
122            map.insert("total".to_string(), serde_json::Value::from(self.total));
123        }
124        map.insert(
125            "data".to_string(),
126            serde_json::Value::from(self.data.to_string()),
127        );
128        if next.is_empty() {
129            map.insert("next".to_string(), serde_json::Value::Null);
130        } else {
131            map.insert("next".to_string(), serde_json::Value::from(next));
132        }
133
134        serde_json::Value::from(map)
135    }
136}
137
138#[cfg(test)]
139mod data_frame_tests {
140    use super::*;
141
142    #[test]
143    fn test_data_frame() {
144        let data_frame = DataFrame {
145            kind: 6,
146            hash: Some(1),
147            index: Some(1),
148            total: Some(1),
149            data: Buffer::from_vec(vec![1]),
150            next: Some(vec![
151                Cid::try_from(
152                    vec![
153                        1, 113, 18, 32, 56, 148, 167, 251, 237, 117, 200, 226, 181, 134, 79, 115,
154                        131, 220, 232, 143, 20, 67, 224, 179, 48, 130, 197, 123, 226, 85, 85, 56,
155                        38, 84, 106, 225,
156                    ]
157                    .as_slice(),
158                )
159                .unwrap(),
160            ]),
161        };
162        let json = data_frame.to_json();
163
164        let wanted_json = serde_json::json!({
165            "kind": 6,
166            "hash": "1",
167            "index": 1,
168            "total": 1,
169            "data": Buffer::from_vec(vec![1]).to_string(),
170            "next": [
171               {
172                "/": "bafyreibysst7x3lvzdrllbspoob5z2epcrb6bmzqqlcxxysvku4cmvdk4e"
173               }
174            ]
175        });
176
177        assert_eq!(json, wanted_json);
178    }
179
180    #[test]
181    fn test_decoding() {
182        {
183            let raw = vec![
184                134, 6, 59, 70, 48, 192, 168, 213, 38, 83, 193, 1, 2, 70, 32, 119, 111, 114, 108,
185                100, 128,
186            ];
187            let as_json_raw = serde_json::json!({"kind":6,"hash":"13388989860809387070","index":1,"total":2,"data":"IHdvcmxk","next":null});
188
189            let data_frame = DataFrame::from_bytes(raw).unwrap();
190            let as_json = data_frame.to_json();
191            assert_eq!(as_json, as_json_raw);
192        }
193        {
194            let raw = vec![
195                134, 6, 27, 72, 172, 245, 101, 152, 189, 52, 248, 24, 26, 24, 28, 74, 178, 79, 233,
196                101, 240, 6, 201, 17, 9, 14, 128,
197            ];
198            let as_json_raw = serde_json::json!({"kind":6,"hash":"5236830283428082936","index":26,"total":28,"data":"sk/pZfAGyREJDg==","next":null});
199
200            let data_frame = DataFrame::from_bytes(raw).unwrap();
201            let as_json = data_frame.to_json();
202            assert_eq!(as_json, as_json_raw);
203        }
204        {
205            let raw = vec![
206                134, 6, 27, 72, 172, 245, 101, 152, 189, 52, 248, 22, 24, 28, 74, 111, 237, 179,
207                173, 165, 39, 99, 171, 113, 233, 133, 216, 42, 88, 37, 0, 1, 113, 18, 32, 122, 71,
208                2, 134, 225, 132, 61, 186, 162, 255, 184, 29, 48, 1, 138, 64, 232, 195, 187, 20, 2,
209                107, 96, 133, 253, 99, 212, 159, 214, 235, 31, 176, 216, 42, 88, 37, 0, 1, 113, 18,
210                32, 28, 140, 185, 170, 59, 82, 138, 35, 215, 213, 58, 142, 227, 82, 31, 146, 35,
211                230, 167, 145, 243, 214, 187, 136, 224, 31, 202, 225, 146, 245, 229, 198, 216, 42,
212                88, 37, 0, 1, 113, 18, 32, 107, 199, 31, 114, 114, 251, 65, 56, 222, 108, 243, 54,
213                182, 63, 194, 178, 61, 197, 69, 4, 128, 71, 62, 116, 222, 43, 105, 250, 14, 182,
214                175, 60, 216, 42, 88, 37, 0, 1, 113, 18, 32, 87, 50, 255, 0, 149, 48, 182, 80, 100,
215                55, 160, 92, 192, 112, 136, 95, 186, 77, 166, 159, 244, 11, 211, 12, 111, 235, 187,
216                124, 29, 52, 146, 102, 216, 42, 88, 37, 0, 1, 113, 18, 32, 81, 216, 114, 215, 30,
217                122, 54, 226, 139, 196, 54, 28, 133, 44, 128, 91, 199, 16, 47, 41, 137, 190, 214,
218                97, 150, 108, 65, 242, 217, 51, 49, 79,
219            ];
220            let as_json_raw = serde_json::json!({"kind":6,"hash":"5236830283428082936","index":22,"total":28,"data":"b+2zraUnY6tx6Q==","next":[{"/":"bafyreid2i4binymehw5kf75yduyadcsa5db3wfacnnqil7ld2sp5n2y7wa"},{"/":"bafyreia4rs42uo2srir5pvj2r3rveh4septkpept225yrya7zlqzf5pfyy"},{"/":"bafyreidly4pxe4x3ie4n43htg23d7qvshxcukbeai47hjxrlnh5a5nvphq"},{"/":"bafyreicxgl7qbfjqwzigin5altahbcc7xjg2nh7ubpjqy37lxn6b2nesmy"},{"/":"bafyreicr3bznoht2g3rixrbwdscszac3y4ic6kmjx3lgdftmihznsmzrj4"}]});
221
222            let data_frame = DataFrame::from_bytes(raw).unwrap();
223            let as_json = data_frame.to_json();
224            assert_eq!(as_json, as_json_raw);
225        }
226    }
227}
jetstreamer_firehose/dataframe.rs

jetstreamer_firehose/
dataframe.rs
