harmos_sidecar/kit/
instance.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// Sidecar binding of the portable host over the connection frames.
11// A submitted child stays owned by the calling attempt; its handle drives it
12// through child controls and releases it when dropped unsettled.
13
14/* === Modules === */
15
16/* === Re-Exports === */
17
18/* === Imports === */
19
20// Standard Library Imports
21use std::future::Future;
22use std::marker::PhantomData;
23
24// Third Party Imports
25
26// Local Imports
27use super::{CallError, Supervisor};
28use harmos_domain::operations::instance::{Id, Status};
29use harmos_domain::operations::lifecycle::InstanceState;
30use harmos_domain::{Message, StreamSpec};
31use harmos_sdk::operations::{CallContext, ContextError, ContextValue, Host, Instance, OperationDeclaration, Response, WireChildAction};
32
33/* === Constants === */
34
35/* === Macros === */
36
37/* === Type Aliases === */
38
39/* === Structs === */
40
41#[must_use = "dropping an unsettled operation instance requests its cancellation"]
42pub struct OperationInstance<O> {
43    host: Supervisor,
44    id: Id,
45    released: bool,
46    _operation: PhantomData<fn() -> O>,
47}
48
49/* === Enums === */
50
51/* === Traits === */
52
53/* === Implementations === */
54
55impl<O: OperationDeclaration> OperationInstance<O> {
56    pub fn id(&self) -> Id {
57        self.id
58    }
59
60    pub async fn status(&self) -> Result<Status, CallError> {
61        let bytes = self.host.child_control(self.id, WireChildAction::Status).await?;
62        decode_status(&bytes)
63    }
64
65    pub async fn cancel(&self) -> Result<(), CallError> {
66        self.host.child_control(self.id, WireChildAction::Cancel).await?;
67        Ok(())
68    }
69
70    pub async fn join(mut self) -> Result<Response<O::Response>, CallError> {
71        let joined = self.host.child_control(self.id, WireChildAction::Join).await;
72        self.released = true;
73        let bytes = joined?;
74        decode_response::<O>(&bytes)
75    }
76
77    pub async fn cancel_and_join(self) -> Result<Response<O::Response>, CallError> {
78        self.cancel().await?;
79        self.join().await
80    }
81
82    // Keep the operation running without a handle until its submitting
83    // attempt ends, which cancels and settles it like any other child.
84    pub fn detach(mut self) -> Id {
85        self.released = true;
86        self.host.child_control_detached(self.id, WireChildAction::Detach);
87        self.id
88    }
89}
90
91impl<O> Drop for OperationInstance<O> {
92    fn drop(&mut self) {
93        if !self.released {
94            self.host.child_control_detached(self.id, WireChildAction::Release);
95        }
96    }
97}
98
99impl<O: OperationDeclaration> Instance<O> for OperationInstance<O> {
100    type Error = CallError;
101
102    fn id(&self) -> Id {
103        self.id
104    }
105
106    fn status(&self) -> impl Future<Output = Result<Status, CallError>> + Send {
107        OperationInstance::status(self)
108    }
109
110    fn cancel(&self) -> impl Future<Output = Result<(), CallError>> + Send {
111        OperationInstance::cancel(self)
112    }
113
114    fn join(self) -> impl Future<Output = Result<Response<O::Response>, CallError>> + Send {
115        OperationInstance::join(self)
116    }
117
118    fn detach(self) -> Id {
119        OperationInstance::detach(self)
120    }
121}
122
123impl Host for Supervisor {
124    type Error = CallError;
125    type Instance<O: OperationDeclaration> = OperationInstance<O>;
126
127    fn context(&self) -> &CallContext {
128        Supervisor::context(self)
129    }
130
131    fn with_value<T: ContextValue>(&self, value: T) -> Result<Self, ContextError> {
132        Supervisor::with_value(self, value)
133    }
134
135    async fn submit<O: OperationDeclaration>(&self, request: O) -> Result<OperationInstance<O>, CallError> {
136        let bytes = self.child_call::<O>(&request, true).await?;
137        let id = u64::decode(bytes.as_slice()).map_err(|error| CallError::Malformed { method: O::ID, message: error.to_string() })?;
138        Ok(OperationInstance { host: self.clone(), id: Id::new(id), released: false, _operation: PhantomData })
139    }
140
141    // Resident resources live in the host process; a sidecar cannot lend them.
142    async fn submit_with_resources<O: OperationDeclaration<Response = ()>, R: Send + Sync + 'static>(
143        &self, _: O, _: std::sync::Arc<R>,
144    ) -> Result<OperationInstance<O>, CallError> {
145        Err(CallError::LocalResources { method: O::ID })
146    }
147
148    async fn call<O: OperationDeclaration>(&self, request: O) -> Result<Response<O::Response>, CallError> {
149        let bytes = self.child_call::<O>(&request, false).await?;
150        decode_response::<O>(&bytes)
151    }
152
153    // Sequence and timestamp each row as the stream publisher does.
154    async fn publish_on<S: StreamSpec>(&self, row: &S::Row) -> Result<(), CallError>
155    where
156        S::Row: Message,
157    {
158        self.publication::<S>(std::slice::from_ref(row), &[self.now()], None).await.map_err(|error| match error {
159            super::PublishError::Publication(error) => CallError::Publication(error),
160            super::PublishError::Call(error) => error,
161        })
162    }
163}
164
165/* === Public Functions === */
166
167/* === Private Functions === */
168
169fn decode_response<O: OperationDeclaration>(bytes: &[u8]) -> Result<Response<O::Response>, CallError> {
170    let response = O::Response::decode(bytes).map_err(|error| CallError::Malformed { method: O::ID, message: error.to_string() })?;
171    Ok(Response::new(response))
172}
173
174fn decode_status(bytes: &[u8]) -> Result<Status, CallError> {
175    let malformed = |message: String| CallError::Malformed { method: harmos_sdk::operations::CHILD_METHOD, message };
176    let state = InstanceState::decode(bytes).map_err(|error| malformed(error.to_string()))?;
177    Status::try_from(&state).map_err(|error| malformed(error.to_string()))
178}
179
180/* === Tests === */
harmos_sidecar/kit/instance.rs

harmos_sidecar/kit/
instance.rs
