harmos_runtime/operations/
streaming.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14pub use harmos_sdk::operations::streaming::{StreamInput, StreamItem, StreamSettings};
15
16/* === Imports === */
17
18// Standard Library Imports
19use std::sync::Arc;
20// Third Party Imports
21// Local Imports
22use crate::{Error, Result};
23use harmos_sdk::operations::streaming as portable;
24
25/* === Constants === */
26
27/* === Macros === */
28
29/* === Type Aliases === */
30
31pub type StreamOutput<T> = portable::StreamOutput<T>;
32pub type StreamPermit<'a, T> = portable::StreamPermit<'a, T>;
33/* === Structs === */
34
35/* === Enums === */
36
37/* === Traits === */
38
39/* === Implementations === */
40
41impl From<portable::Error> for Error {
42    fn from(error: portable::Error) -> Self {
43        match error {
44            portable::Error::Settings => Self::InvalidInput {
45                field: "operation stream limits",
46                reason: "capacity exceeds the semaphore claim limit",
47            },
48            portable::Error::Capacity => Self::ResourceExhausted { resource: "operation stream bytes" },
49            portable::Error::Construction => Self::InvalidInput {
50                field: "operation stream construction bound",
51                reason: "constructed value exceeds its reserved bytes",
52            },
53            portable::Error::Closed => Self::NotReady { domain: "operation stream input" },
54            portable::Error::Ownership => Self::InvalidInput {
55                field: "operation stream payload",
56                reason: "original owned payload cannot be moved into a request",
57            },
58            portable::Error::Codec(source) => Self::TaskFailed {
59                domain: "operation stream decode",
60                source: Arc::new(source),
61            },
62        }
63    }
64}
65/* === Public Functions === */
66
67pub fn stream<T>(settings: StreamSettings) -> Result<(StreamOutput<T>, StreamInput<T>)> {
68    portable::channel(settings).map_err(Into::into)
69}
70/* === Private Functions === */
71
72/* === Tests === */
harmos_runtime/operations/streaming.rs

harmos_runtime/operations/
streaming.rs
