saluki_core/components/sources/
context.rs

1use std::sync::Arc;
2
3use saluki_common::sync::shutdown::ShutdownHandle;
4
5use crate::accounting::ComponentRegistry;
6use crate::health::Health;
7use crate::{
8    components::ComponentContext,
9    topology::{EventsDispatcher, TopologyContext},
10};
11
12struct SourceContextInner {
13    topology_context: TopologyContext,
14    component_context: ComponentContext,
15    component_registry: ComponentRegistry,
16    dispatcher: EventsDispatcher,
17}
18
19/// Source context.
20pub struct SourceContext {
21    shutdown_handle: Option<ShutdownHandle>,
22    health_handle: Option<Health>,
23    inner: Arc<SourceContextInner>,
24}
25
26impl SourceContext {
27    /// Creates a new `SourceContext`.
28    pub fn new(
29        topology_context: &TopologyContext, component_context: &ComponentContext,
30        component_registry: ComponentRegistry, health_handle: Health, dispatcher: EventsDispatcher,
31    ) -> Self {
32        Self {
33            shutdown_handle: None,
34            health_handle: Some(health_handle),
35            inner: Arc::new(SourceContextInner {
36                topology_context: topology_context.clone(),
37                component_context: component_context.clone(),
38                component_registry,
39                dispatcher,
40            }),
41        }
42    }
43
44    /// Installs the shutdown handle for this source context.
45    ///
46    /// Called once by the runtime, before the component runs, with the shutdown signal of the component's dedicated
47    /// supervisor.
48    pub(crate) fn set_shutdown_handle(&mut self, shutdown_handle: ShutdownHandle) {
49        self.shutdown_handle = Some(shutdown_handle);
50    }
51
52    /// Consumes the shutdown handle of this source context.
53    ///
54    /// # Panics
55    ///
56    /// Panics if the shutdown handle has already been taken.
57    pub fn take_shutdown_handle(&mut self) -> ShutdownHandle {
58        self.shutdown_handle.take().expect("shutdown handle already taken")
59    }
60
61    /// Consumes the health handle of this source context.
62    ///
63    /// # Panics
64    ///
65    /// Panics if the health handle has already been taken.
66    pub fn take_health_handle(&mut self) -> Health {
67        self.health_handle.take().expect("health handle already taken")
68    }
69
70    /// Returns a reference to the topology context.
71    pub fn topology_context(&self) -> &TopologyContext {
72        &self.inner.topology_context
73    }
74
75    /// Returns a reference to the component context.
76    pub fn component_context(&self) -> &ComponentContext {
77        &self.inner.component_context
78    }
79
80    /// Returns a reference to the component registry.
81    pub fn component_registry(&self) -> &ComponentRegistry {
82        &self.inner.component_registry
83    }
84
85    /// Returns a reference to the events dispatcher.
86    pub fn dispatcher(&self) -> &EventsDispatcher {
87        &self.inner.dispatcher
88    }
89}
90
91impl Clone for SourceContext {
92    fn clone(&self) -> Self {
93        Self {
94            shutdown_handle: None,
95            health_handle: None,
96            inner: self.inner.clone(),
97        }
98    }
99}
saluki_core/components/sources/context.rs

saluki_core/components/sources/
context.rs
