harmos_domain/model/storage/
writes.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17
18// Third Party Imports
19
20// Local Imports
21use super::{ArrowIpc, SchemaKey, StorageCursor};
22use crate::{Identifier, MemorySize};
23
24/* === Constants === */
25
26/* === Macros === */
27
28crate::model::storage::codec::product!(DataBatch {
29    sequence: crate::model::storage::codec::Unsigned => 1,
30    destination: crate::model::storage::codec::Id => 2,
31    schema: crate::model::storage::codec::Nested<SchemaKey> => 3,
32    data: crate::model::storage::codec::Nested<ArrowIpc> => 4,
33});
34crate::model::storage::codec::product!(BatchReceipt {
35    sequence: crate::model::storage::codec::Unsigned => 1,
36    cursor: crate::model::storage::codec::Nested<StorageCursor> => 2,
37});
38crate::model::storage::codec::product!(WriteCompletion {
39    batches: crate::model::storage::codec::Unsigned => 1,
40    cursor: crate::model::storage::codec::Nested<StorageCursor> => 2,
41});
42crate::model::storage::codec::variants!(WriteReceipt {
43    0, 1, "accepted": Self::Accepted(receipt) => {
44        receipt: crate::model::storage::codec::Nested<BatchReceipt> => 1,
45    } => Self::Accepted(receipt);
46    1, 2, "durable": Self::Durable(receipt) => {
47        receipt: crate::model::storage::codec::Nested<BatchReceipt> => 1,
48    } => Self::Durable(receipt);
49});
50
51/* === Type Aliases === */
52
53/* === Structs === */
54
55#[derive(Clone, Debug)]
56pub struct DataBatch {
57    pub sequence: u64,
58    pub destination: Identifier,
59    pub schema: SchemaKey,
60    pub data: ArrowIpc,
61}
62
63#[derive(Clone, Debug, PartialEq, Eq)]
64pub struct BatchReceipt {
65    pub sequence: u64,
66    pub cursor: StorageCursor,
67}
68
69#[derive(Clone, Debug, PartialEq, Eq)]
70pub struct WriteCompletion {
71    pub batches: u64,
72    pub cursor: StorageCursor,
73}
74
75/* === Enums === */
76
77#[derive(Clone, Debug, PartialEq, Eq)]
78pub enum WriteReceipt {
79    Accepted(BatchReceipt),
80    Durable(BatchReceipt),
81}
82
83/* === Traits === */
84
85/* === Implementations === */
86
87impl MemorySize for DataBatch {
88    fn memory_size(&self) -> usize {
89        std::mem::size_of::<Self>()
90            .saturating_add(self.destination.memory_size().saturating_sub(std::mem::size_of_val(&self.destination)))
91            .saturating_add(self.schema.memory_size().saturating_sub(std::mem::size_of_val(&self.schema)))
92            .saturating_add(self.data.memory_size().saturating_sub(std::mem::size_of_val(&self.data)))
93    }
94}
95
96impl MemorySize for BatchReceipt {
97    fn memory_size(&self) -> usize {
98        std::mem::size_of::<Self>().saturating_add(self.cursor.memory_size().saturating_sub(std::mem::size_of_val(&self.cursor)))
99    }
100}
101
102impl MemorySize for WriteReceipt {
103    fn memory_size(&self) -> usize {
104        let receipt = match self {
105            Self::Accepted(receipt) | Self::Durable(receipt) => receipt,
106        };
107        std::mem::size_of::<Self>().saturating_add(receipt.memory_size().saturating_sub(std::mem::size_of::<BatchReceipt>()))
108    }
109}
110
111impl MemorySize for WriteCompletion {
112    fn memory_size(&self) -> usize {
113        std::mem::size_of::<Self>().saturating_add(self.cursor.memory_size().saturating_sub(std::mem::size_of_val(&self.cursor)))
114    }
115}
116
117/* === Public Functions === */
118
119/* === Private Functions === */
120
121/* === Tests === */
harmos_domain/model/storage/writes.rs

harmos_domain/model/storage/
writes.rs
