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, RecordInput};
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    #[tracing::instrument(target = "harmos::storage", name = "storage.push", level = "debug", skip_all)]
64    pub async fn push(&mut self, record: RecordInput) -> Result<()> {
65        if self.refused {
66            return Err(limit());
67        }
68        self.refused = true;
69        let result = self.permit.push(record, self.limits).await;
70        self.refused = result.is_err();
71        result.map_err(Into::into)
72    }
73
74    pub(crate) fn check(&self, maximum: std::num::NonZeroUsize) -> Result<()> {
75        if self.refused {
76            return Err(limit());
77        }
78        for entry in self.permit.projected() {
79            if encoded_size(entry)? > maximum.get() {
80                return Err(Error::InvalidInput { field: "record projection", reason: "one entry exceeds max_batch_bytes" });
81            }
82        }
83        Ok(())
84    }
85}
86
87/* === Public Functions === */
88
89/* === Private Functions === */
90
91fn limit() -> Error {
92    Error::InvalidInput { field: "record projection", reason: "capture limit exceeded" }
93}
94
95/* === Tests === */
harmos_runtime/storage/capture.rs

harmos_runtime/storage/
capture.rs
