harmos_domain/operations/
context.rs1use std::any::{Any, TypeId, type_name};
18use std::collections::{BTreeMap, HashMap};
19use std::sync::Arc;
20use crate::{Message, MessageError};
24
25pub const PORTABLE_VALUES: usize = 64;
28
29pub const PORTABLE_BYTES: usize = 64 * 1024;
30
31#[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#[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
72pub trait ContextValue: Message + Send + Sync + 'static {
75 const KEY: &'static str;
76}
77
78impl 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 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 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