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

harmos_runtime/operations/
streaming.rs
