harmos_domain/operations/
context.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
17use std::any::{Any, TypeId, type_name};
18use std::collections::{BTreeMap, HashMap};
19use std::sync::Arc;
20// Third Party Imports
21
22// Local Imports
23use crate::{Message, MessageError};
24
25/* === Constants === */
26
27pub const PORTABLE_VALUES: usize = 64;
28
29pub const PORTABLE_BYTES: usize = 64 * 1024;
30
31/* === Macros === */
32
33/* === Type Aliases === */
34
35/* === Structs === */
36
37#[derive(Clone, Default)]
38pub struct CallContext {
39    values: HashMap<TypeId, Arc<dyn Any + Send + Sync>>,
40    portable: BTreeMap<String, Arc<[u8]>>,
41    portable_types: HashMap<&'static str, TypeId>,
42}
43
44#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
45pub struct CallValue {
46    pub key: String,
47    pub payload: Vec<u8>,
48}
49
50/* === Enums === */
51
52#[derive(Debug, thiserror::Error)]
53pub enum ContextError {
54    #[error("call context already carries `{name}`")]
55    Duplicate { name: String },
56
57    #[error("call context does not carry `{name}`")]
58    Missing { name: String },
59
60    #[error("call context value `{key}` is invalid: {source}")]
61    Decode {
62        key: String,
63
64        #[source]
65        source: MessageError,
66    },
67
68    #[error("call context portable values exceed {PORTABLE_VALUES} entries or {PORTABLE_BYTES} bytes")]
69    Budget,
70}
71
72/* === Traits === */
73
74pub trait ContextValue: Message + Send + Sync + 'static {
75    const KEY: &'static str;
76}
77
78/* === Implementations === */
79
80impl CallContext {
81    pub fn insert<T: Send + Sync + 'static>(&mut self, value: T) -> Result<(), ContextError> {
82        let key = TypeId::of::<T>();
83        if self.values.contains_key(&key) {
84            return Err(ContextError::Duplicate { name: type_name::<T>().to_owned() });
85        }
86        self.values.insert(key, Arc::new(value));
87        Ok(())
88    }
89
90    pub fn insert_portable<T: ContextValue>(&mut self, value: T) -> Result<(), ContextError> {
91        if self.portable.contains_key(T::KEY) {
92            return Err(ContextError::Duplicate { name: T::KEY.to_owned() });
93        }
94        let payload = value.encode();
95        let bytes = self.portable.iter().map(|(key, value)| key.len() + value.len()).sum::<usize>() + T::KEY.len() + payload.len();
96        if self.portable.len() >= PORTABLE_VALUES || bytes > PORTABLE_BYTES {
97            return Err(ContextError::Budget);
98        }
99        self.insert(value)?;
100        self.portable.insert(T::KEY.to_owned(), payload.into());
101        self.portable_types.insert(T::KEY, TypeId::of::<T>());
102        Ok(())
103    }
104
105    // Derive a child context with a replaced portable value. Local-only values
106    // retain their original allocation and cannot be replaced by this method.
107    pub fn with_value<T: ContextValue>(&self, value: T) -> Result<Self, ContextError> {
108        let kind = TypeId::of::<T>();
109        if self.values.contains_key(&kind) && self.portable_types.get(T::KEY) != Some(&kind) {
110            return Err(ContextError::Duplicate { name: type_name::<T>().to_owned() });
111        }
112        let replacement = CallValue {
113            key: T::KEY.to_owned(),
114            payload: value.encode(),
115        };
116        let mut child = self.with_values(vec![replacement])?;
117        child.values.insert(kind, Arc::new(value));
118        child.portable_types.insert(T::KEY, kind);
119        Ok(child)
120    }
121
122    // Overlay portable values without changing trusted local-only values.
123    // Replaced typed projections are invalidated; read wire values with `value`.
124    pub fn with_values(&self, values: Vec<CallValue>) -> Result<Self, ContextError> {
125        let replacements = Self::from_portable(values)?;
126        let retained = self.portable.iter().filter(|(key, _)| !replacements.portable.contains_key(*key));
127        let mut entries = replacements.portable.len();
128        let mut bytes =
129            replacements.portable.iter().map(|(key, value)| key.len().saturating_add(value.len())).fold(0usize, usize::saturating_add);
130        for (key, value) in retained {
131            entries = entries.saturating_add(1);
132            bytes = bytes.saturating_add(key.len()).saturating_add(value.len());
133        }
134        if entries > PORTABLE_VALUES || bytes > PORTABLE_BYTES {
135            return Err(ContextError::Budget);
136        }
137        let mut child = self.clone();
138        for (key, payload) in replacements.portable {
139            if let Some(kind) = child.portable_types.remove(key.as_str()) {
140                child.values.remove(&kind);
141            }
142            child.portable.insert(key, payload);
143        }
144        Ok(child)
145    }
146
147    pub fn from_portable(values: Vec<CallValue>) -> Result<Self, ContextError> {
148        let bytes = values.iter().map(|value| value.key.len() + value.payload.len()).sum::<usize>();
149        if values.len() > PORTABLE_VALUES || bytes > PORTABLE_BYTES {
150            return Err(ContextError::Budget);
151        }
152        let mut context = Self::default();
153        for value in values {
154            if context.portable.contains_key(&value.key) {
155                return Err(ContextError::Duplicate { name: value.key });
156            }
157            context.portable.insert(value.key, value.payload.into());
158        }
159        Ok(context)
160    }
161
162    pub fn get<T: Send + Sync + 'static>(&self) -> Option<&T> {
163        self.values.get(&TypeId::of::<T>())?.downcast_ref()
164    }
165
166    pub fn require<T: Send + Sync + 'static>(&self) -> Result<&T, ContextError> {
167        self.get().ok_or_else(|| ContextError::Missing { name: type_name::<T>().to_owned() })
168    }
169
170    pub fn value<T: ContextValue>(&self) -> Result<Option<T>, ContextError> {
171        let Some(payload) = self.portable.get(T::KEY) else {
172            return Ok(None);
173        };
174        T::decode(payload).map(Some).map_err(|source| ContextError::Decode { key: T::KEY.to_owned(), source })
175    }
176
177    pub fn portable(&self) -> Vec<CallValue> {
178        self.portable
179            .iter()
180            .map(|(key, payload)| CallValue {
181                key: key.clone(),
182                payload: payload.to_vec(),
183            })
184            .collect()
185    }
186
187    pub fn is_empty(&self) -> bool {
188        self.values.is_empty() && self.portable.is_empty()
189    }
190}
191
192/* === Public Functions === */
193
194/* === Private Functions === */
195
196/* === Tests === */
harmos_domain/operations/context.rs

harmos_domain/operations/
context.rs
