harmos_runtime/storage/
capture.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::manager::encoded_size;
22use super::state::Permit;
23use super::{ReadLimits, RecordWrite};
24use crate::{Error, Result, state::Envelope};
25use std::{future::Future, pin::Pin};
26
27/* === Constants === */
28
29/* === Macros === */
30
31/* === Type Aliases === */
32
33/* === Structs === */
34
35pub struct ProjectedRecords<'a> {
36    permit: &'a mut Permit,
37    limits: ReadLimits,
38    refused: bool,
39}
40
41pub(crate) struct Projection {
42    pub(crate) limits: ReadLimits,
43    pub(crate) handler: Box<dyn RecordProjection>,
44}
45
46/* === Enums === */
47
48/* === Traits === */
49
50pub trait RecordProjection: Send + Sync + 'static {
51    fn project<'a>(
52        &'a self, record: &'a Envelope<&serde_json::Value>, output: &'a mut ProjectedRecords<'_>,
53    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'a>>;
54}
55
56/* === Implementations === */
57
58impl<'a> ProjectedRecords<'a> {
59    pub(crate) fn new(permit: &'a mut Permit, limits: ReadLimits) -> Self {
60        Self { permit, limits, refused: false }
61    }
62
63    pub async fn push(&mut self, record: RecordWrite<'_>) -> Result<()> {
64        if self.refused {
65            return Err(limit());
66        }
67        self.refused = true;
68        let result = self.permit.push(record, self.limits).await;
69        self.refused = result.is_err();
70        result.map_err(Into::into)
71    }
72
73    pub(crate) fn check(&self, maximum: std::num::NonZeroUsize) -> Result<()> {
74        if self.refused {
75            return Err(limit());
76        }
77        for entry in self.permit.projected() {
78            if encoded_size(entry)? > maximum.get() {
79                return Err(Error::InvalidInput { field: "record projection", reason: "one entry exceeds max_batch_bytes" });
80            }
81        }
82        Ok(())
83    }
84}
85
86/* === Public Functions === */
87
88/* === Private Functions === */
89
90fn limit() -> Error {
91    Error::InvalidInput { field: "record projection", reason: "capture limit exceeded" }
92}
93
94/* === Tests === */
harmos_runtime/storage/capture.rs

harmos_runtime/storage/
capture.rs
