harmos_runtime/operations/
session.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::sync::{
18    Arc,
19    atomic::{AtomicBool, Ordering},
20};
21
22// Third Party Imports
23
24// Local Imports
25use super::{
26    OperationGeneration, OperationInstance, OperationInstanceId, OperationManager, OperationStatus, StreamInput, StreamItem, StreamOutput,
27    StreamSettings, stream,
28};
29use crate::operations::OperationDeclaration;
30use crate::supervision::instance::{Instance, Phase};
31use crate::{Error, Refusal, Result};
32use harmos_domain::MemorySize;
33
34/* === Constants === */
35
36/* === Macros === */
37
38/* === Type Aliases === */
39
40pub type Upload<O> = Session<O, SessionInput<<<O as OperationDeclaration>::Input as crate::operations::Direction>::Item>, ()>;
41pub type Download<O> = Session<O, (), SessionOutput<<<O as OperationDeclaration>::Output as crate::operations::Direction>::Item>>;
42pub type Exchange<O> = Session<
43    O,
44    SessionInput<<<O as OperationDeclaration>::Input as crate::operations::Direction>::Item>,
45    SessionOutput<<<O as OperationDeclaration>::Output as crate::operations::Direction>::Item>,
46>;
47
48/* === Structs === */
49
50#[must_use = "dropping a streaming session requests cancellation"]
51pub struct Session<O: OperationDeclaration, I = (), R = ()> {
52    input: I,
53    output: R,
54    completion: Completion<O>,
55}
56
57pub struct SessionInput<T> {
58    sender: Option<StreamOutput<T>>,
59    ends: Arc<Ends>,
60    control: Arc<Instance>,
61}
62
63pub struct SessionOutput<T> {
64    receiver: StreamInput<T>,
65    ends: Arc<Ends>,
66    control: Arc<Instance>,
67}
68
69#[must_use = "join completion or explicitly cancel and settle"]
70pub struct Completion<O: OperationDeclaration> {
71    instance: OperationInstance<O>,
72    ends: Arc<Ends>,
73    delivered: bool,
74}
75
76// Channel Progress Only; Execution And Cleanup Remain On Instance.
77struct Ends {
78    input: AtomicBool,
79    output: AtomicBool,
80}
81
82/* === Enums === */
83
84/* === Traits === */
85
86/* === Implementations === */
87
88impl<O: OperationDeclaration, I, R> Session<O, I, R> {
89    pub fn id(&self) -> OperationInstanceId {
90        self.completion.id()
91    }
92
93    pub fn generation(&self) -> OperationGeneration {
94        self.completion.generation()
95    }
96
97    pub fn status(&self) -> OperationStatus {
98        self.completion.status()
99    }
100
101    pub fn cancel(&self) {
102        self.completion.cancel();
103    }
104
105    pub fn split(self) -> (I, R, Completion<O>) {
106        (self.input, self.output, self.completion)
107    }
108
109    pub async fn join(self) -> Result<crate::operations::Response<O::Response>> {
110        let (input, output, completion) = self.split();
111        drop(input);
112        drop(output);
113        completion.join().await
114    }
115
116    pub async fn cancel_and_settle(self) -> Result<()> {
117        self.cancel();
118        let (input, output, completion) = self.split();
119        drop(input);
120        drop(output);
121        completion.cancel_and_settle().await
122    }
123}
124
125impl<O: OperationDeclaration, T, R> Session<O, SessionInput<T>, R> {
126    pub fn finish_input(&mut self) {
127        self.input.finish_input();
128    }
129}
130impl<O: OperationDeclaration, T: MemorySize, R> Session<O, SessionInput<T>, R> {
131    pub async fn send(&self, item: T) -> Result<()> {
132        self.input.send(item).await
133    }
134
135    pub async fn send_with(&self, bytes: usize, make: impl FnOnce() -> Result<T>) -> Result<()> {
136        self.input.send_with(bytes, make).await
137    }
138}
139impl<O: OperationDeclaration, I, T> Session<O, I, SessionOutput<T>> {
140    pub async fn next(&mut self) -> Result<Option<StreamItem<T>>> {
141        self.output.next().await
142    }
143}
144
145impl<T> SessionInput<T> {
146    pub fn finish_input(&mut self) {
147        self.ends.input.store(true, Ordering::Release);
148        self.sender.take();
149    }
150}
151impl<T: MemorySize> SessionInput<T> {
152    pub async fn send(&self, item: T) -> Result<()> {
153        self.send_with(item.memory_size(), || Ok(item)).await
154    }
155
156    pub async fn send_with(&self, bytes: usize, make: impl FnOnce() -> Result<T>) -> Result<()> {
157        let sender = self.sender.as_ref().ok_or(Error::NotReady { domain: "operation input closed" })?;
158        tokio::select! {
159            biased;
160            error = ended(&self.control) => Err(error),
161            result = sender.send_with(bytes, make) => result,
162        }
163    }
164}
165impl<T> Drop for SessionInput<T> {
166    fn drop(&mut self) {
167        if !self.ends.input.load(Ordering::Acquire) {
168            cancel_unsettled(&self.control);
169        }
170    }
171}
172
173impl<T> SessionOutput<T> {
174    pub async fn next(&mut self) -> Result<Option<StreamItem<T>>> {
175        if self.ends.output.load(Ordering::Acquire) {
176            return Ok(None);
177        }
178        let mut status = self.control.settlement();
179        loop {
180            let current = status.borrow_and_update().clone();
181            if current.cancellation_requested() {
182                self.receiver.close();
183                return Err(Refusal::Cancelled.into());
184            }
185            if current.outcome().is_some() {
186                self.receiver.close();
187                let item = self.receiver.try_recv().ok();
188                if item.is_none() {
189                    self.ends.output.store(true, Ordering::Release);
190                }
191                return Ok(item);
192            }
193            tokio::select! {
194                biased;
195                item = self.receiver.recv() => {
196                    if item.is_none() { self.ends.output.store(true, Ordering::Release); }
197                    return Ok(item);
198                }
199                changed = status.changed() => {
200                    if changed.is_err() { return Err(Refusal::Unavailable.into()); }
201                }
202            }
203        }
204    }
205}
206impl<T> Drop for SessionOutput<T> {
207    fn drop(&mut self) {
208        if !self.ends.output.load(Ordering::Acquire) {
209            cancel_unsettled(&self.control);
210        }
211    }
212}
213
214impl<O: OperationDeclaration> Completion<O> {
215    pub fn id(&self) -> OperationInstanceId {
216        self.instance.id()
217    }
218
219    pub fn generation(&self) -> OperationGeneration {
220        self.instance.generation()
221    }
222
223    pub fn status(&self) -> OperationStatus {
224        self.instance.status()
225    }
226
227    pub fn cancel(&self) {
228        self.instance.cancel();
229    }
230
231    pub async fn settle(&self) -> Result<()> {
232        self.instance.settle().await
233    }
234
235    pub async fn join(mut self) -> Result<crate::operations::Response<O::Response>> {
236        if !self.ends.input.load(Ordering::Acquire) || !self.ends.output.load(Ordering::Acquire) {
237            return Err(Error::InvalidInput {
238                field: "operation stream completion",
239                reason: "finish input and drain output before joining; use cancel_and_settle to abandon",
240            });
241        }
242        let result = (&mut self.instance).await;
243        self.delivered = true;
244        result
245    }
246
247    pub async fn cancel_and_settle(mut self) -> Result<()> {
248        self.instance.cancel();
249        let result = self.instance.settle().await;
250        self.delivered = true;
251        result
252    }
253}
254impl<O: OperationDeclaration> Drop for Completion<O> {
255    fn drop(&mut self) {
256        if !self.delivered {
257            cancel_unsettled(&self.instance.control);
258        }
259    }
260}
261
262impl OperationManager {
263    pub fn upload<O: crate::operations::OperationDeclaration>(&self, request: O, settings: StreamSettings) -> Result<Upload<O>> {
264        upload(request, settings, |request, input| self.submit_endpoints::<O, _>(request, input))
265    }
266
267    pub fn download<O: crate::operations::OperationDeclaration>(&self, request: O, settings: StreamSettings) -> Result<Download<O>> {
268        download(request, settings, |request, output| self.submit_endpoints::<O, _>(request, output))
269    }
270
271    pub fn exchange<O: crate::operations::OperationDeclaration>(
272        &self, request: O, input: StreamSettings, output: StreamSettings,
273    ) -> Result<Exchange<O>> {
274        exchange(request, input, output, |request, endpoints| self.submit_endpoints::<O, _>(request, endpoints))
275    }
276}
277
278impl super::Context<'_> {
279    pub fn upload<O: crate::operations::OperationDeclaration>(&self, request: O, settings: StreamSettings) -> Result<Upload<O>> {
280        upload(request, settings, |request, input| self.submit_endpoints::<O, _>(request, input))
281    }
282
283    pub fn download<O: crate::operations::OperationDeclaration>(&self, request: O, settings: StreamSettings) -> Result<Download<O>> {
284        download(request, settings, |request, output| self.submit_endpoints::<O, _>(request, output))
285    }
286
287    pub fn exchange<O: crate::operations::OperationDeclaration>(
288        &self, request: O, input: StreamSettings, output: StreamSettings,
289    ) -> Result<Exchange<O>> {
290        exchange(request, input, output, |request, endpoints| self.submit_endpoints::<O, _>(request, endpoints))
291    }
292}
293
294#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
295impl super::Caller {
296    pub fn upload<O: crate::operations::OperationDeclaration>(&self, request: O, settings: StreamSettings) -> Result<Upload<O>> {
297        upload(request, settings, |request, input| self.submit_endpoints::<O, _>(request, input))
298    }
299
300    pub fn download<O: crate::operations::OperationDeclaration>(&self, request: O, settings: StreamSettings) -> Result<Download<O>> {
301        download(request, settings, |request, output| self.submit_endpoints::<O, _>(request, output))
302    }
303
304    pub fn exchange<O: crate::operations::OperationDeclaration>(
305        &self, request: O, input: StreamSettings, output: StreamSettings,
306    ) -> Result<Exchange<O>> {
307        exchange(request, input, output, |request, endpoints| self.submit_endpoints::<O, _>(request, endpoints))
308    }
309}
310
311/* === Public Functions === */
312
313/* === Private Functions === */
314
315fn upload<O: crate::operations::OperationDeclaration>(
316    request: O, settings: StreamSettings,
317    submit: impl FnOnce(
318        O,
319        StreamInput<<<O as OperationDeclaration>::Input as crate::operations::Direction>::Item>,
320    ) -> Result<OperationInstance<O>>,
321) -> Result<Upload<O>> {
322    let (sender, input) = stream(settings)?;
323    let instance = submit(request, input)?;
324    let ends = Arc::new(Ends { input: AtomicBool::new(false), output: AtomicBool::new(true) });
325    Ok(Session {
326        input: SessionInput { sender: Some(sender), ends: ends.clone(), control: instance.control.clone() },
327        output: (),
328        completion: Completion { instance, ends, delivered: false },
329    })
330}
331fn download<O: crate::operations::OperationDeclaration>(
332    request: O, settings: StreamSettings,
333    submit: impl FnOnce(
334        O,
335        StreamOutput<<<O as OperationDeclaration>::Output as crate::operations::Direction>::Item>,
336    ) -> Result<OperationInstance<O>>,
337) -> Result<Download<O>> {
338    let (output, receiver) = stream(settings)?;
339    let instance = submit(request, output)?;
340    let ends = Arc::new(Ends { input: AtomicBool::new(true), output: AtomicBool::new(false) });
341    Ok(Session {
342        input: (),
343        output: SessionOutput { receiver, ends: ends.clone(), control: instance.control.clone() },
344        completion: Completion { instance, ends, delivered: false },
345    })
346}
347fn exchange<O: crate::operations::OperationDeclaration>(
348    request: O, input: StreamSettings, output: StreamSettings,
349    submit: impl FnOnce(
350        O,
351        (
352            StreamInput<<<O as OperationDeclaration>::Input as crate::operations::Direction>::Item>,
353            StreamOutput<<<O as OperationDeclaration>::Output as crate::operations::Direction>::Item>,
354        ),
355    ) -> Result<OperationInstance<O>>,
356) -> Result<Exchange<O>> {
357    let (sender, input) = stream(input)?;
358    let (output, receiver) = stream(output)?;
359    let instance = submit(request, (input, output))?;
360    let ends = Arc::new(Ends { input: AtomicBool::new(false), output: AtomicBool::new(false) });
361    Ok(Session {
362        input: SessionInput { sender: Some(sender), ends: ends.clone(), control: instance.control.clone() },
363        output: SessionOutput { receiver, ends: ends.clone(), control: instance.control.clone() },
364        completion: Completion { instance, ends, delivered: false },
365    })
366}
367fn cancel_unsettled(control: &Instance) {
368    if control.status().phase() != Phase::Settled {
369        control.cancel();
370    }
371}
372async fn ended(control: &Instance) -> Error {
373    let mut status = control.settlement();
374    loop {
375        let current = status.borrow_and_update().clone();
376        if current.cancellation_requested() {
377            return Refusal::Cancelled.into();
378        }
379        if current.outcome().is_some() {
380            return Error::NotReady { domain: "operation completed" };
381        }
382        if status.changed().await.is_err() {
383            return Refusal::Unavailable.into();
384        }
385    }
386}
387
388/* === Tests === */
harmos_runtime/operations/session.rs

harmos_runtime/operations/
session.rs
