harmos_sdk/operations/
operation.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::{TypeId, type_name};
18use std::future::Future;
19use std::marker::PhantomData;
20use std::sync::Arc;
21
22// Third Party Imports
23
24// Local Imports
25#[cfg(feature = "streaming")]
26use super::Stream;
27use super::calls::{Calls, InvocationError, Target};
28use super::{
29    CallContext, Carrier, EndpointSet, Executable, ExecutionContext, FailureKind, Invokes, NoStream, OperationDeclaration, Requires,
30    Response,
31};
32
33/* === Constants === */
34
35/* === Macros === */
36
37/* === Type Aliases === */
38
39/* === Structs === */
40
41// One Execution Resource An Operation Declares With `resources = [..]`.
42#[derive(Clone, Copy, Debug)]
43pub struct Resource {
44    id: fn() -> TypeId,
45    name: fn() -> &'static str,
46}
47
48// One Attempt Of An Authored Operation: Call Metadata, Declared Resources,
49// Typed Stream Endpoints And Declared Child Operations.
50pub struct Context<'a, O: Streams> {
51    execution: &'a ExecutionContext,
52    streams: &'a O::StreamEndpoints,
53}
54
55// The Carrier Behind Every Authored Operation; Each Attempt Runs A Clone Of
56// The Request So A Retry Keeps The Original.
57pub struct Authored<O>(PhantomData<fn() -> O>);
58
59/* === Enums === */
60
61/* === Traits === */
62
63// An Operation Whose Struct Is Its Request Message.
64pub trait Operation: OperationDeclaration<Request = Self> + crate::Portable + Streams + Dependent + Clone + Send + 'static {
65    type Error: std::error::Error + Send + Sync + 'static;
66
67    fn run(self, context: Context<'_, Self>) -> impl Future<Output = Result<Response<Self::Response>, Self::Error>> + Send;
68
69    fn classify(_error: &Self::Error) -> FailureKind {
70        FailureKind::Permanent
71    }
72}
73
74// The Resources `#[harmos::operation]` Records For Registration Verification.
75pub trait Dependent {
76    const RESOURCES: &'static [Resource];
77}
78
79// The Endpoint Set Derived From A Declaration's `input` And `output`.
80pub trait Streams {
81    type StreamEndpoints: EndpointSet;
82}
83
84pub trait Shape {
85    type StreamEndpoints: EndpointSet;
86}
87
88/* === Implementations === */
89
90impl Resource {
91    pub const fn of<T: Send + Sync + 'static>() -> Self {
92        Self {
93            id: TypeId::of::<T>,
94            name: type_name::<T>,
95        }
96    }
97
98    pub fn id(&self) -> TypeId {
99        (self.id)()
100    }
101
102    pub fn name(&self) -> &'static str {
103        (self.name)()
104    }
105}
106
107impl<O: Streams> Context<'_, O> {
108    pub fn call(&self) -> &CallContext {
109        &self.execution.call
110    }
111
112    // Registration Verified Every Declared Resource Before This Attempt Began.
113    pub fn resource<T: Send + Sync + 'static>(&self) -> Arc<T>
114    where
115        O: Requires<T>,
116    {
117        self.execution.resource::<T>().unwrap_or_else(|error| panic!("declared resource unavailable after verification: {error}"))
118    }
119
120    pub fn streams(&self) -> &O::StreamEndpoints {
121        self.streams
122    }
123
124    // Invoke A Declared Operation With This Attempt's Call Metadata; The Callee's Policy Applies.
125    pub fn operation<T>(&self, request: T) -> impl Future<Output = Result<Response<T::Response>, InvocationError>> + Send + use<'_, O, T>
126    where
127        O: Invokes<T>,
128        T: OperationDeclaration<Request = T> + Send,
129    {
130        let calls = self.execution.resource::<Calls>().map_err(|_| InvocationError::Unavailable { operation: T::ID });
131        async move {
132            let response = calls?.0.call(Target::of::<T>(), Box::new(request)).await?;
133            response.restore::<T::Response>().ok_or(InvocationError::Response { operation: T::ID })
134        }
135    }
136}
137
138impl<O: Operation> Carrier for Authored<O> {
139    type Declaration = O;
140    type Error = O::Error;
141    type StreamEndpoints = O::StreamEndpoints;
142
143    async fn bind(_: ExecutionContext) -> Result<Self, O::Error> {
144        Ok(Self(PhantomData))
145    }
146
147    async fn attempt(
148        &mut self, context: ExecutionContext, request: &mut O, endpoints: &mut O::StreamEndpoints,
149    ) -> Result<Response<O::Response>, O::Error> {
150        request.clone().run(Context { execution: &context, streams: endpoints }).await
151    }
152
153    fn transferred(&self) -> bool {
154        false
155    }
156
157    fn classify(error: &O::Error) -> FailureKind {
158        O::classify(error)
159    }
160}
161
162impl<O: Operation> Executable for O {
163    type Carrier = Authored<O>;
164    const RESOURCES: &'static [Resource] = <O as Dependent>::RESOURCES;
165}
166
167impl<O: OperationDeclaration> Streams for O
168where
169    (O::Input, O::Output): Shape,
170{
171    type StreamEndpoints = <(O::Input, O::Output) as Shape>::StreamEndpoints;
172}
173
174impl Shape for (NoStream, NoStream) {
175    type StreamEndpoints = ();
176}
177
178#[cfg(feature = "streaming")]
179impl<T: crate::Portable> Shape for (Stream<T>, NoStream) {
180    type StreamEndpoints = super::StreamInput<T>;
181}
182
183#[cfg(feature = "streaming")]
184impl<T: crate::Portable> Shape for (NoStream, Stream<T>) {
185    type StreamEndpoints = super::StreamOutput<T>;
186}
187
188#[cfg(feature = "streaming")]
189impl<I: crate::Portable, T: crate::Portable> Shape for (Stream<I>, Stream<T>) {
190    type StreamEndpoints = (super::StreamInput<I>, super::StreamOutput<T>);
191}
192
193/* === Public Functions === */
194
195/* === Private Functions === */
196
197/* === Tests === */
198
199#[cfg(test)]
200mod tests {
201    use super::*;
202    use crate::operations::{Dependencies, RegistrationBatch};
203
204    #[crate::operation(harmos_path = crate, resources = [u64], id = "test.increment", response = u64,
205        policy = crate::OperationPolicy {
206            timeout: std::time::Duration::from_secs(1),
207            attempt_timeout: std::time::Duration::from_secs(1),
208            retries: crate::RetryPolicy::Never,
209        },
210    )]
211    struct Increment {
212        value: u64,
213    }
214
215    impl Operation for Increment {
216        type Error = std::convert::Infallible;
217        async fn run(mut self, ctx: Context<'_, Self>) -> Result<Response<u64>, Self::Error> {
218            self.value += *ctx.resource::<u64>();
219            Ok(self.value.into())
220        }
221    }
222
223    fn ready<T>(future: impl Future<Output = T>) -> T {
224        let mut future = std::pin::pin!(future);
225        let mut context = std::task::Context::from_waker(std::task::Waker::noop());
226        match future.as_mut().poll(&mut context) {
227            std::task::Poll::Ready(value) => value,
228            std::task::Poll::Pending => panic!("synchronous operation pending"),
229        }
230    }
231
232    #[test]
233    fn declared_resources_and_attempt_clones() {
234        let missing = RegistrationBatch::new(Dependencies::new()).operation::<Increment>();
235        assert!(missing.operations[0].verify_resources(&[]).is_err());
236        assert!(missing.operations[0].verify_resources(&[TypeId::of::<u64>()]).is_ok());
237        let mut dependencies = Dependencies::new();
238        dependencies.insert(Arc::new(2_u64)).unwrap();
239        let batch = RegistrationBatch::new(dependencies).operation::<Increment>();
240        let factory = &batch.operations[0];
241        factory.verify_resources(&[]).unwrap();
242        let mut instance = factory.prepare_native(Box::new(Increment { value: 5 }), Box::new(())).unwrap();
243        let context = ExecutionContext::new(Dependencies::new(), CallContext::default());
244        ready(instance.initialize(context.clone())).unwrap();
245        for _ in 0..2 {
246            let answer = ready(instance.run(context.clone())).unwrap().restore::<u64>().unwrap();
247            assert_eq!(*answer, 7);
248        }
249    }
250}
harmos_sdk/operations/operation.rs

harmos_sdk/operations/
operation.rs
