hydro_lang/
runtime_context.rs

1//! Interfaces to access DFIR runtime metadata from Hydro programs.
2
3use dfir_rs::scheduled::context::Context;
4use quote::quote;
5use stageleft::runtime_support::{FreeVariableWithContextWithProps, QuoteTokens};
6
7use crate::live_collections::ContextWithLocation;
8
9/// Exposes the DFIR [`Context`] inside quoted code.
10pub static RUNTIME_CONTEXT: RuntimeContext<'static> = RuntimeContext { _private: &() };
11
12/// A handle to DFIR [`Context`] which can be used inside quoted code.
13///
14/// When spliced, it turns into a reference to [`Context`].
15///
16/// The exposed type is `&Context<()>` (the "no metrics" typestate), since whether the actual
17/// runtime context tracks metrics is only decided at deploy time (e.g. by telemetry sidecars).
18/// `Context<()>` only exposes the methods available on *every* `Context<M>` (e.g.
19/// [`Context::current_tick`]), so quoted code type-checked against it will also compile against
20/// the real context regardless of which metrics typestate it ends up with.
21#[derive(Clone, Copy)]
22pub struct RuntimeContext<'a> {
23    _private: &'a (),
24}
25
26impl<'a, Ctx> FreeVariableWithContextWithProps<Ctx, ()> for RuntimeContext<'a>
27where
28    Ctx: ContextWithLocation<'a>,
29{
30    type O = &'a Context<()>;
31
32    fn to_tokens(self, _ctx: &Ctx) -> (QuoteTokens, ()) {
33        (
34            QuoteTokens {
35                prelude: None,
36                expr: Some(quote!(&context)),
37            },
38            (),
39        )
40    }
41}
42
43#[cfg(test)]
44mod tests {
45    use futures::StreamExt;
46    use hydro_deploy::Deployment;
47    use stageleft::q;
48
49    use super::RUNTIME_CONTEXT;
50    use crate::compile::builder::FlowBuilder;
51    use crate::location::Location;
52
53    struct P1 {}
54
55    #[tokio::test]
56    async fn runtime_context() {
57        let mut deployment = Deployment::new();
58
59        let mut flow = FlowBuilder::new();
60        let node = flow.process::<P1>();
61        let external = flow.external::<()>();
62
63        let out_port = node
64            .source_iter(q!(0..5))
65            .map(q!(|v| (v, RUNTIME_CONTEXT.current_tick().0)))
66            .send_bincode_external(&external);
67
68        let nodes = flow
69            .with_process(&node, deployment.Localhost())
70            .with_external(&external, deployment.Localhost())
71            .deploy(&mut deployment);
72
73        deployment.deploy().await.unwrap();
74
75        let mut external_out = nodes.connect(out_port).await;
76
77        deployment.start().await.unwrap();
78
79        for i in 0..5 {
80            assert_eq!(external_out.next().await.unwrap(), (i, 0));
81        }
82    }
83}
hydro_lang/runtime_context.rs

hydro_lang/
runtime_context.rs
