harmos_domain/model/storage/
data.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12pub mod codec;
13
14/* === Re-Exports === */
15
16/* === Imports === */
17
18// Standard Library Imports
19use std::time::SystemTime as Timestamp;
20
21// Third Party Imports
22use serde::{Deserialize, Serialize};
23
24// Local Imports
25use crate::model::storage::{SchemaKey, StorageCursor};
26use crate::{Identifier, MemorySize};
27
28/* === Constants === */
29
30/* === Macros === */
31
32/* === Type Aliases === */
33
34/* === Structs === */
35
36#[derive(Clone, Debug, PartialEq, Eq)]
37pub struct RecordInput {
38    pub schema: SchemaKey,
39    pub scope: Option<Identifier>,
40    pub item: Identifier,
41    pub source: Option<Identifier>,
42    pub source_time: Option<Timestamp>,
43    pub payload: Vec<u8>,
44}
45
46#[derive(Clone, Debug, PartialEq, Eq)]
47pub struct StreamInput {
48    pub record: RecordInput,
49    pub stream: Identifier,
50    pub sequence: u64,
51}
52
53#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
54#[serde(deny_unknown_fields)]
55pub struct StoredRecord<T = Vec<u8>> {
56    pub position: StorageCursor,
57    pub schema: SchemaKey,
58    pub scope: Option<Identifier>,
59    pub item: Identifier,
60    pub source: Option<Identifier>,
61
62    #[serde(default, with = "crate::model::storage::time::optional")]
63    pub source_time: Option<Timestamp>,
64
65    pub payload: T,
66}
67
68#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
69#[serde(deny_unknown_fields)]
70pub struct StoredRow<T = Vec<u8>> {
71    pub record: StoredRecord<T>,
72    pub stream: Identifier,
73    pub sequence: u64,
74}
75
76#[derive(Clone, Copy, Debug, Serialize)]
77pub struct RecordWrite<'a> {
78    pub schema: &'a SchemaKey,
79    pub scope: Option<&'a Identifier>,
80    pub item: &'a Identifier,
81    pub source: Option<&'a Identifier>,
82    pub source_time: Option<Timestamp>,
83    pub payload: &'a [u8],
84}
85
86#[derive(Clone, Copy, Debug, Serialize)]
87pub struct StreamWrite<'a> {
88    pub stream: &'a Identifier,
89    pub sequence: u64,
90    pub record: RecordWrite<'a>,
91}
92
93/* === Enums === */
94
95/* === Traits === */
96
97/* === Implementations === */
98
99impl MemorySize for RecordInput {
100    fn memory_size(&self) -> usize {
101        std::mem::size_of::<Self>() + self.schema.memory_size() - std::mem::size_of::<SchemaKey>() + self.scope.memory_size()
102            - std::mem::size_of::<Option<Identifier>>()
103            + self.item.memory_size()
104            - std::mem::size_of::<Identifier>()
105            + self.source.memory_size()
106            - std::mem::size_of::<Option<Identifier>>()
107            + self.payload.capacity()
108    }
109}
110impl MemorySize for StreamInput {
111    fn memory_size(&self) -> usize {
112        std::mem::size_of::<Self>() + self.record.memory_size() - std::mem::size_of::<RecordInput>() + self.stream.memory_size()
113            - std::mem::size_of::<Identifier>()
114    }
115}
116
117impl RecordInput {
118    pub fn borrowed(&self) -> RecordWrite<'_> {
119        RecordWrite {
120            schema: &self.schema,
121            scope: self.scope.as_ref(),
122            item: &self.item,
123            source: self.source.as_ref(),
124            source_time: self.source_time,
125            payload: &self.payload,
126        }
127    }
128}
129
130impl StreamInput {
131    pub fn borrowed(&self) -> StreamWrite<'_> {
132        StreamWrite { record: self.record.borrowed(), stream: &self.stream, sequence: self.sequence }
133    }
134}
135
136impl RecordWrite<'_> {
137    pub fn stored(&self, position: StorageCursor) -> StoredRecord {
138        StoredRecord {
139            position,
140            schema: self.schema.clone(),
141            scope: self.scope.cloned(),
142            item: self.item.clone(),
143            source: self.source.cloned(),
144            source_time: self.source_time,
145            payload: self.payload.to_vec(),
146        }
147    }
148}
149
150impl<T: MemorySize> MemorySize for StoredRecord<T> {
151    fn memory_size(&self) -> usize {
152        let dynamic = [
153            self.position.memory_size().saturating_sub(std::mem::size_of::<StorageCursor>()),
154            self.schema.memory_size().saturating_sub(std::mem::size_of::<SchemaKey>()),
155            self.scope.memory_size().saturating_sub(std::mem::size_of::<Option<Identifier>>()),
156            self.item.memory_size().saturating_sub(std::mem::size_of::<Identifier>()),
157            self.source.memory_size().saturating_sub(std::mem::size_of::<Option<Identifier>>()),
158            self.payload.memory_size().saturating_sub(std::mem::size_of::<T>()),
159        ];
160        dynamic.into_iter().fold(std::mem::size_of::<Self>(), usize::saturating_add)
161    }
162}
163
164impl<T: MemorySize> MemorySize for StoredRow<T> {
165    fn memory_size(&self) -> usize {
166        std::mem::size_of::<Self>()
167            .saturating_add(self.stream.memory_size().saturating_sub(std::mem::size_of::<Identifier>()))
168            .saturating_add(self.record.memory_size().saturating_sub(std::mem::size_of::<StoredRecord<T>>()))
169    }
170}
171
172/* === Public Functions === */
173
174/* === Private Functions === */
175
176/* === Tests === */
harmos_domain/model/storage/data.rs

harmos_domain/model/storage/
data.rs
