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

use dfir_rs::scheduled::context::Context;
use proc_macro2::TokenStream;
use quote::quote;
use stageleft::runtime_support::FreeVariableWithContext;

use crate::Location;

pub static RUNTIME_CONTEXT: RuntimeContext = RuntimeContext { _private: &() };

#[derive(Clone, Copy)]
pub struct RuntimeContext<'a> {
    _private: &'a (),
}

impl<'a, L: Location<'a>> FreeVariableWithContext<L> for RuntimeContext<'a> {
    type O = &'a Context;

    fn to_tokens(self, _ctx: &L) -> (Option<TokenStream>, Option<TokenStream>) {
        (None, Some(quote!(&context)))
    }
}

#[cfg(test)]
mod tests {
    use dfir_rs::futures::StreamExt;
    use hydro_deploy::Deployment;

    use crate::*;

    struct P1 {}

    #[tokio::test]
    async fn runtime_context() {
        let mut deployment = Deployment::new();

        let flow = FlowBuilder::new();
        let node = flow.process::<P1>();
        let external = flow.external_process::<()>();

        let out_port = node
            .source_iter(q!(0..5))
            .map(q!(|v| (v, RUNTIME_CONTEXT.current_tick().0)))
            .send_bincode_external(&external);

        let nodes = flow
            .with_process(&node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut external_out = nodes.connect_source_bincode(out_port).await;

        deployment.start().await.unwrap();

        for i in 0..5 {
            assert_eq!(external_out.next().await.unwrap(), (i, 0));
        }
    }
}