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 stored(self, position: StorageCursor) -> StoredRecord {
119        StoredRecord {
120            position,
121            schema: self.schema,
122            scope: self.scope,
123            item: self.item,
124            source: self.source,
125            source_time: self.source_time,
126            payload: self.payload,
127        }
128    }
129
130    pub fn borrowed(&self) -> RecordWrite<'_> {
131        RecordWrite {
132            schema: &self.schema,
133            scope: self.scope.as_ref(),
134            item: &self.item,
135            source: self.source.as_ref(),
136            source_time: self.source_time,
137            payload: &self.payload,
138        }
139    }
140}
141
142impl StreamInput {
143    pub fn borrowed(&self) -> StreamWrite<'_> {
144        StreamWrite { record: self.record.borrowed(), stream: &self.stream, sequence: self.sequence }
145    }
146}
147
148impl<T: MemorySize> MemorySize for StoredRecord<T> {
149    fn memory_size(&self) -> usize {
150        let dynamic = [
151            self.position.memory_size().saturating_sub(std::mem::size_of::<StorageCursor>()),
152            self.schema.memory_size().saturating_sub(std::mem::size_of::<SchemaKey>()),
153            self.scope.memory_size().saturating_sub(std::mem::size_of::<Option<Identifier>>()),
154            self.item.memory_size().saturating_sub(std::mem::size_of::<Identifier>()),
155            self.source.memory_size().saturating_sub(std::mem::size_of::<Option<Identifier>>()),
156            self.payload.memory_size().saturating_sub(std::mem::size_of::<T>()),
157        ];
158        dynamic.into_iter().fold(std::mem::size_of::<Self>(), usize::saturating_add)
159    }
160}
161
162impl<T: MemorySize> MemorySize for StoredRow<T> {
163    fn memory_size(&self) -> usize {
164        std::mem::size_of::<Self>()
165            .saturating_add(self.stream.memory_size().saturating_sub(std::mem::size_of::<Identifier>()))
166            .saturating_add(self.record.memory_size().saturating_sub(std::mem::size_of::<StoredRecord<T>>()))
167    }
168}
169
170/* === Public Functions === */
171
172/* === Private Functions === */
173
174/* === Tests === */
harmos_domain/model/storage/data.rs

harmos_domain/model/storage/
data.rs
