harmos-sdk 0.1.6

Harmos SDK for authoring plugins and operations over the Harmos runtime.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

/* ***
 *
 * Harmos
 *
 * SPDX-License-Identifier: Apache-2.0
 *
 * ***
 */

/* === Modules === */

/* === Re-Exports === */

/* === Imports === */

// Standard Library Imports
use std::any::Any;
// Third Party Imports

// Local Imports
use crate::grpc::{self, DecodeInput};
#[cfg(feature = "streaming")]
use crate::operations::Direction;
use crate::operations::OperationDeclaration;
use crate::{MessageError, Portable};

/* === Constants === */

/* === Macros === */

/* === Type Aliases === */

type Decode = fn(&[u8], &mut dyn Any) -> Result<(), MessageError>;
type Encode = fn(&dyn Any) -> Result<Vec<u8>, MessageError>;

#[cfg(feature = "streaming")]
type Prepare = fn(&[u8]) -> Result<(Vec<u8>, usize), MessageError>;

/* === Structs === */

#[derive(Clone, Copy)]
pub(super) struct Codec {
    pub(super) decode: Decode,
    pub(super) encode: Encode,

    #[cfg(feature = "streaming")]
    pub(super) input: Option<ItemCodec>,

    #[cfg(feature = "streaming")]
    pub(super) output: Option<ItemCodec>,
}

#[cfg(feature = "streaming")]
#[derive(Clone, Copy)]
pub(super) struct ItemCodec {
    pub(super) item_type: std::any::TypeId,
    pub(super) prepare: Prepare,
    pub(super) decode: Decode,
    pub(super) encode: Encode,
}

/* === Enums === */

/* === Traits === */

/* === Implementations === */

impl Codec {
    // The One Protobuf Projection Of A Declaration: Request, Response And The
    // Item Codec Of Each Direction It Streams.
    pub(crate) fn of<O: OperationDeclaration>() -> Self {
        Self {
            decode: decode::<O>,
            encode: encode::<O::Response>,
            #[cfg(feature = "streaming")]
            input: direction::<O::Input>(),
            #[cfg(feature = "streaming")]
            output: direction::<O::Output>(),
        }
    }
}

/* === Public Functions === */

/* === Private Functions === */

#[cfg(feature = "streaming")]
fn direction<D: Direction>() -> Option<ItemCodec> {
    D::STREAMED.then(item::<D::Item>)
}

#[cfg(feature = "streaming")]
fn item<T: Portable>() -> ItemCodec {
    ItemCodec {
        item_type: std::any::TypeId::of::<T>(),
        prepare: |payload| {
            let prepared = grpc::prepare::<T>(DecodeInput::new(payload))?;
            let memory = prepared.memory();
            Ok((prepared.into_postcard(), memory))
        },
        decode: decode_postcard::<T>,
        encode: encode::<T>,
    }
}

fn decode<T: Portable>(payload: &[u8], target: &mut dyn Any) -> Result<(), MessageError> {
    let target =
        target.downcast_mut::<Option<T>>().ok_or_else(|| MessageError::borrowed("protobuf request type differs from operation"))?;
    *target = Some(grpc::decode::<T>(DecodeInput::new(payload))?);
    Ok(())
}

#[cfg(feature = "streaming")]
#[tracing::instrument(target = "harmos::grpc", name = "grpc.postcard.decode", level = "trace", skip_all, fields(bytes = payload.len()))]
fn decode_postcard<T: Portable>(payload: &[u8], target: &mut dyn Any) -> Result<(), MessageError> {
    let target =
        target.downcast_mut::<Option<T>>().ok_or_else(|| MessageError::borrowed("protobuf stream item type differs from operation"))?;
    *target = Some(T::decode(payload)?);
    Ok(())
}

fn encode<T: Portable>(value: &dyn Any) -> Result<Vec<u8>, MessageError> {
    let value = value.downcast_ref::<T>().ok_or_else(|| MessageError::borrowed("protobuf response type differs from operation"))?;
    let mut bytes = Vec::new();
    grpc::encode(value, &mut bytes)?;
    Ok(bytes)
}

/* === Tests === */