titanrt 0.7.0

Typed reactive runtime for real-time systems
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

use anyhow::anyhow;
use bytes::Bytes;
use tonic::{Code, Status, metadata::MetadataMap};

use crate::connector::features::shared::events::StreamEventInner;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GrpcEventKind {
    StreamConnected,
    StreamDisconnected,
    StreamItem,
    UnaryResponse,
    BadCommand,
}

#[derive(Debug, Clone)]
pub struct GrpcEvent {
    kind: GrpcEventKind,
    code: Code,
    err_msg: Option<String>,
    metadata: MetadataMap,
    body: Option<Bytes>,
}

impl GrpcEvent {
    pub fn from_ok() -> Self {
        Self {
            kind: GrpcEventKind::StreamConnected,
            code: Code::Ok,
            err_msg: None,
            metadata: MetadataMap::new(),
            body: None,
        }
    }

    pub fn from_ok_unary(resp: tonic::Response<Bytes>) -> Self {
        let (metadata, body, _trailing) = resp.into_parts();
        Self {
            kind: GrpcEventKind::UnaryResponse,
            code: Code::Ok,
            err_msg: None,
            metadata,
            body: Some(body),
        }
    }

    pub fn from_ok_stream_item(body: Bytes) -> Self {
        Self {
            kind: GrpcEventKind::StreamItem,
            code: Code::Ok,
            err_msg: None,
            metadata: MetadataMap::new(),
            body: Some(body),
        }
    }

    pub fn from_ok_stream_close(metadata: MetadataMap) -> Self {
        let status = Status::aborted("grpc streaming aborted".to_string());
        Self {
            kind: GrpcEventKind::StreamDisconnected,
            code: status.code(),
            err_msg: Some(status.message().to_string()),
            metadata,
            body: None,
        }
    }

    pub fn from_status(kind: GrpcEventKind, st: Status) -> Self {
        Self {
            kind,
            code: st.code(),
            err_msg: Some(st.message().to_string()),
            metadata: st.metadata().clone(),
            body: None,
        }
    }

    pub fn kind(&self) -> GrpcEventKind {
        self.kind
    }

    pub fn metadata(&self) -> &MetadataMap {
        &self.metadata
    }

    pub fn decode_as<T: prost::Message + Default>(&self) -> anyhow::Result<T> {
        if let Some(ref b) = self.body {
            T::decode(b.as_ref()).map_err(|e| anyhow!("Prost decode error: {e}"))
        } else {
            Err(anyhow!("No body to decode"))
        }
    }
}

impl StreamEventInner for GrpcEvent {
    type Body = Bytes;
    type Err = String;
    type Code = Code;

    fn status(&self) -> Option<&Self::Code> {
        Some(&self.code)
    }

    fn is_ok(&self) -> bool {
        self.code == Code::Ok
    }

    fn error(&self) -> Option<&Self::Err> {
        self.err_msg.as_ref()
    }

    fn body(&self) -> Option<&Self::Body> {
        self.body.as_ref()
    }

    fn into_body(self) -> Option<Self::Body> {
        self.body
    }
}