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

use anyhow::{Context, anyhow};

use crate::connector::{
    BaseConnector, EventTxType,
    features::composite::{
        CompositeConnector,
        stream::{
            StreamKind, StreamSpec, StreamWrapper,
            event::{StreamEventContext, StreamEventParsed},
            hooks::{grpc_hook, http_hook, ws_hook},
        },
    },
};

impl<E: StreamEventParsed> CompositeConnector<E> {
    pub fn spawn_stream(
        &self,
        spec: &StreamSpec,
        ctx: &StreamEventContext<E>,
    ) -> anyhow::Result<StreamWrapper<E>>
    where
        E: StreamEventParsed,
    {
        match spec.kind {
            StreamKind::Http => {
                let descriptor = spec
                    .maybe_http(&ctx)
                    .context("failed to build HTTP descriptor")?;

                let stream = self
                    .with_http(|conn| {
                        conn.spawn_stream(
                            descriptor,
                            EventTxType::External(self.event_tx.clone()),
                            http_hook,
                        )
                    })?
                    .ok_or_else(|| anyhow!("reqwest connector is not configured"))?;

                Ok(StreamWrapper::Http { stream })
            }
            StreamKind::Grpc => {
                let descriptor = spec
                    .maybe_grpc(&ctx)
                    .context("failed to build gRPC descriptor")?;

                let stream = self
                    .with_grpc(|conn| {
                        conn.spawn_stream(
                            descriptor,
                            EventTxType::External(self.event_tx.clone()),
                            grpc_hook,
                        )
                    })?
                    .ok_or_else(|| anyhow!("tonic connector is not configured"))?;

                Ok(StreamWrapper::Grpc { stream })
            }
            StreamKind::Ws => {
                let descriptor = spec
                    .maybe_ws(&ctx)
                    .context("failed to build WebSocket descriptor")?;

                let stream = self
                    .with_websocket(|conn| {
                        conn.spawn_stream(
                            descriptor,
                            EventTxType::External(self.event_tx.clone()),
                            ws_hook,
                        )
                    })?
                    .ok_or_else(|| anyhow!("websocket connector is not configured"))?;

                Ok(StreamWrapper::Ws { stream })
            }
        }
    }
}