somatize_data/
ipc.rs

1//! The codec for frames: Arrow IPC, in both directions.
2
3use crate::Frame;
4use somatize_core::Value;
5use somatize_core::{Codec, CodecError, as_written, written_down};
6
7/// Writes frames down so they can be kept or sent, and reads them back.
8///
9/// The **second implementor of `Codec`, and from another crate** — the first was
10/// `python/`'s registry of `dump`/`load` pairs — which is what keeps the hole a
11/// hole. A unit struct because the format is the format and there is nothing to
12/// configure; whoever runs the engine hands it in, and a store and a wire get
13/// the same bytes because what a frame weighs has one answer.
14///
15/// It refuses an opaque that is not a frame rather than guessing. A graph
16/// carrying tensors **and** rows wants both codecs, which is what `python/`'s
17/// does: a frame comes here, anything else goes to its registry.
18pub struct Ipc;
19
20impl Codec for Ipc {
21    fn packed(&self, value: &Value) -> Result<Value, CodecError> {
22        Ok(match value {
23            Value::Opaque(_) => {
24                let frame = Frame::of(value).ok_or_else(|| {
25                    CodecError::new(
26                        "this opaque value is not a frame, and Arrow IPC is all this codec \
27                         knows how to write down",
28                    )
29                })?;
30                written_down(
31                    Frame::KIND,
32                    frame.written().map_err(|e| CodecError::new(e.message()))?,
33                )
34            }
35            Value::Map(pairs) => Value::map(
36                pairs
37                    .iter()
38                    .map(|(key, value)| Ok((key.clone(), self.packed(value)?)))
39                    .collect::<Result<Vec<_>, CodecError>>()?,
40            ),
41            Value::List(items) => Value::list(
42                items
43                    .iter()
44                    .map(|item| self.packed(item))
45                    .collect::<Result<Vec<_>, CodecError>>()?,
46            ),
47            other => other.clone(),
48        })
49    }
50
51    fn unpacked(&self, value: &Value) -> Result<Value, CodecError> {
52        if let Some((kind, bytes)) = as_written(value) {
53            // Somebody else's kind is left exactly as it arrived: it is not
54            // ours to read and it is not ours to lose, and the process that
55            // does know it may be one hop further on.
56            if kind != Frame::KIND {
57                return Ok(value.clone());
58            }
59            return Ok(Frame::read(bytes)
60                .map_err(|e| CodecError::new(e.message()))?
61                .value());
62        }
63        Ok(match value {
64            Value::Map(pairs) => Value::map(
65                pairs
66                    .iter()
67                    .map(|(key, value)| Ok((key.clone(), self.unpacked(value)?)))
68                    .collect::<Result<Vec<_>, CodecError>>()?,
69            ),
70            Value::List(items) => Value::list(
71                items
72                    .iter()
73                    .map(|item| self.unpacked(item))
74                    .collect::<Result<Vec<_>, CodecError>>()?,
75            ),
76            other => other.clone(),
77        })
78    }
79}
somatize_data/ipc.rs

somatize_data/
ipc.rs
