harmos_operations/storage/
write_streams.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::IO_POLICY;
22use harmos_domain::Identifier;
23use harmos_domain::MemorySize;
24use harmos_domain::model::storage::codec::{Id, Optional, product};
25use harmos_domain::model::storage::{DataBatch, WriteCompletion, WriteReceipt};
26use harmos_domain::operations::ReplaySafety;
27#[cfg(feature = "runtime")]
28use harmos_sdk::operations::access::GrantAccess;
29#[cfg(feature = "runtime")]
30use harmos_storage::{ErasedCatalog, StorageManager};
31
32/* === Constants === */
33
34/* === Macros === */
35
36product!(WriteStreams {
37    scope: Optional<Id> => 1,
38});
39
40/* === Type Aliases === */
41
42/* === Structs === */
43
44// Admit Streamed Arrow Batches Of Stream Rows Into One Optional Scope.
45#[harmos::operation(
46    harmos_path = crate,
47    resources = [StorageManager<ErasedCatalog>, GrantAccess],
48    codec = manual,
49    id = "storage.write_streams",
50    response = WriteCompletion,
51    policy = IO_POLICY,
52    input = DataBatch,
53    output = WriteReceipt,
54    visibility = Shared,
55    replay = ReplaySafety::Unknown,
56    grpc(package = "harmos.storage.v1", service =  "Storage", method =  "WriteStreams"),
57)]
58#[derive(Clone, Debug, PartialEq, Eq)]
59pub struct WriteStreams {
60    pub scope: Option<Identifier>,
61}
62
63/* === Enums === */
64
65/* === Traits === */
66
67/* === Implementations === */
68
69impl MemorySize for WriteStreams {
70    fn memory_size(&self) -> usize {
71        std::mem::size_of::<Self>().saturating_add(self.scope.memory_size().saturating_sub(std::mem::size_of_val(&self.scope)))
72    }
73}
74
75#[cfg(feature = "runtime")]
76mod execution {
77    use super::*;
78    use crate::storage::exchange::{WriteKind, exchange};
79
80    // Standard Library Imports
81    // Third Party Imports
82    // Local Imports
83    use crate::{Error, Result};
84    use harmos_sdk::operations::{Operation, OperationDeclaration};
85    use harmos_storage::writes::WriteCompletion;
86
87    impl Operation for super::WriteStreams {
88        type Error = Error;
89
90        async fn run(self, context: harmos_sdk::Context<'_, Self>) -> Result<harmos_sdk::operations::Response<WriteCompletion>> {
91            let storage = context.resource::<harmos_storage::StorageManager<harmos_storage::ErasedCatalog>>();
92            let grants = context.resource::<harmos_sdk::operations::access::GrantAccess>();
93            let endpoints = context.streams();
94
95            let host = grants;
96            let ctx = &host.0;
97            let completion = exchange(&storage, ctx, self.scope.as_ref(), endpoints, WriteKind::Streams, WriteStreams::ID).await?;
98            Ok(completion.into())
99        }
100
101        fn classify(error: &Error) -> harmos_sdk::operations::FailureKind {
102            crate::storage::classify(error)
103        }
104    }
105}
106
107/* === Public Functions === */
108
109/* === Private Functions === */
110
111/* === Tests === */
harmos_operations/storage/write_streams.rs

harmos_operations/storage/
write_streams.rs
