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::{Message, StreamSpec};
30use harmos_operations::lifecycle::InstanceState;
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    async fn publish_on<S: StreamSpec>(&self, row: &S::Row) -> Result<(), CallError>
125    where
126        S::Row: Message,
127    {
128        self.publication::<S>(&[row.encode()], &[self.now()], None).await.map_err(|error| match error {
129            super::PublishError::Publication(error) => CallError::Publication(error),
130            super::PublishError::Call(error) => error,
131        })
132    }
133    type Error = CallError;
134    type Instance<O: OperationDeclaration> = OperationInstance<O>;
135
136    fn context(&self) -> &CallContext {
137        Supervisor::context(self)
138    }
139
140    fn with_value<T: ContextValue>(&self, value: T) -> Result<Self, ContextError> {
141        Supervisor::with_value(self, value)
142    }
143
144    async fn submit<O: OperationDeclaration>(&self, request: O::Request) -> Result<OperationInstance<O>, CallError> {
145        let bytes = self.child_call::<O>(&request, true).await?;
146        let id = u64::decode(bytes.as_slice()).map_err(|error| CallError::Malformed {
147            method: O::ID,
148            message: error.to_string(),
149        })?;
150        Ok(OperationInstance {
151            host: self.clone(),
152            id: Id::new(id),
153            released: false,
154            _operation: PhantomData,
155        })
156    }
157
158    async fn submit_owned<O: OperationDeclaration>(
159        &self, _: O::Request, _: harmos_sdk::operations::RequestClaims,
160    ) -> Result<OperationInstance<O>, CallError> {
161        Err(CallError::LocalResources { method: O::ID })
162    }
163
164    async fn call<O: OperationDeclaration>(&self, request: O::Request) -> Result<Response<O::Response>, CallError> {
165        let bytes = self.child_call::<O>(&request, false).await?;
166        decode_response::<O>(&bytes)
167    }
168
169    // Sequence and timestamp each row as the stream publisher does.
170}
171
172/* === Public Functions === */
173
174/* === Private Functions === */
175
176fn decode_response<O: OperationDeclaration>(bytes: &[u8]) -> Result<Response<O::Response>, CallError> {
177    let response = O::Response::decode(bytes).map_err(|error| CallError::Malformed {
178        method: O::ID,
179        message: error.to_string(),
180    })?;
181    Ok(Response::new(response))
182}
183
184fn decode_status(bytes: &[u8]) -> Result<Status, CallError> {
185    let malformed = |message: String| CallError::Malformed {
186        method: harmos_sdk::operations::CHILD_METHOD,
187        message,
188    };
189    let state = InstanceState::decode(bytes).map_err(|error| malformed(error.to_string()))?;
190    Status::try_from(&state).map_err(|error| malformed(error.to_string()))
191}
192
193/* === Tests === */
harmos_sidecar/kit/instance.rs

harmos_sidecar/kit/
instance.rs
