hydroflow_plus_deploy 0.9.0

Library for working with hydro_deploy and hydroflow_plus
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

use hydroflow_plus::util::deploy::{
    ConnectedDemux, ConnectedDirect, ConnectedSink, ConnectedSource, ConnectedTagged, DeployPorts,
};
use stageleft::{q, Quoted, RuntimeData};

use crate::HydroflowPlusMeta;

pub fn cluster_members<'a>(of_cluster: usize) -> impl Quoted<'a, &'a Vec<u32>> + Copy + 'a {
    let cli: RuntimeData<&DeployPorts<HydroflowPlusMeta>> =
        RuntimeData::new("__hydroflow_plus_trybuild_cli");
    q!(cli.meta.clusters.get(&of_cluster).unwrap())
}

pub fn cluster_self_id<'a>() -> impl Quoted<'a, u32> + Copy + 'a {
    let cli: RuntimeData<&DeployPorts<HydroflowPlusMeta>> =
        RuntimeData::new("__hydroflow_plus_trybuild_cli");
    q!(cli
        .meta
        .cluster_id
        .expect("Tried to read Cluster ID on a non-cluster node"))
}

pub fn deploy_o2o(p1_port: &str, p2_port: &str) -> (syn::Expr, syn::Expr) {
    let env: RuntimeData<&DeployPorts<HydroflowPlusMeta>> =
        RuntimeData::new("__hydroflow_plus_trybuild_cli");
    (
        {
            q!({
                env.port(p1_port)
                    .connect_local_blocking::<ConnectedDirect>()
                    .into_sink()
            })
            .splice()
        },
        {
            q!({
                env.port(p2_port)
                    .connect_local_blocking::<ConnectedDirect>()
                    .into_source()
            })
            .splice()
        },
    )
}

pub fn deploy_o2m(p1_port: &str, c2_port: &str) -> (syn::Expr, syn::Expr) {
    let env: RuntimeData<&DeployPorts<HydroflowPlusMeta>> =
        RuntimeData::new("__hydroflow_plus_trybuild_cli");
    (
        {
            q!({
                env.port(p1_port)
                    .connect_local_blocking::<ConnectedDemux<ConnectedDirect>>()
                    .into_sink()
            })
            .splice()
        },
        {
            q!({
                env.port(c2_port)
                    .connect_local_blocking::<ConnectedDirect>()
                    .into_source()
            })
            .splice()
        },
    )
}

pub fn deploy_m2o(c1_port: &str, p2_port: &str) -> (syn::Expr, syn::Expr) {
    let env: RuntimeData<&DeployPorts<HydroflowPlusMeta>> =
        RuntimeData::new("__hydroflow_plus_trybuild_cli");
    (
        {
            q!({
                env.port(c1_port)
                    .connect_local_blocking::<ConnectedDirect>()
                    .into_sink()
            })
            .splice()
        },
        {
            q!({
                env.port(p2_port)
                    .connect_local_blocking::<ConnectedTagged<ConnectedDirect>>()
                    .into_source()
            })
            .splice()
        },
    )
}

pub fn deploy_m2m(c1_port: &str, c2_port: &str) -> (syn::Expr, syn::Expr) {
    let env: RuntimeData<&DeployPorts<HydroflowPlusMeta>> =
        RuntimeData::new("__hydroflow_plus_trybuild_cli");
    (
        {
            q!({
                env.port(c1_port)
                    .connect_local_blocking::<ConnectedDemux<ConnectedDirect>>()
                    .into_sink()
            })
            .splice()
        },
        {
            q!({
                env.port(c2_port)
                    .connect_local_blocking::<ConnectedTagged<ConnectedDirect>>()
                    .into_source()
            })
            .splice()
        },
    )
}