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::runtime::SupervisorHandle;
8use crate::{
9    components::ComponentContext,
10    topology::{EventsDispatcher, TopologyContext},
11};
12
13struct SourceContextInner {
14    topology_context: TopologyContext,
15    component_context: ComponentContext,
16    component_registry: ComponentRegistry,
17    dispatcher: EventsDispatcher,
18    supervisor_handle: SupervisorHandle,
19}
20
21/// Source context.
22pub struct SourceContext {
23    shutdown_handle: Option<ShutdownHandle>,
24    health_handle: Option<Health>,
25    inner: Arc<SourceContextInner>,
26}
27
28impl SourceContext {
29    /// Creates a new `SourceContext`.
30    pub fn new(
31        topology_context: &TopologyContext, component_context: &ComponentContext,
32        component_registry: ComponentRegistry, health_handle: Health, dispatcher: EventsDispatcher,
33        supervisor_handle: SupervisorHandle,
34    ) -> Self {
35        Self {
36            shutdown_handle: None,
37            health_handle: Some(health_handle),
38            inner: Arc::new(SourceContextInner {
39                topology_context: topology_context.clone(),
40                component_context: component_context.clone(),
41                component_registry,
42                dispatcher,
43                supervisor_handle,
44            }),
45        }
46    }
47
48    /// Installs the shutdown handle for this source context.
49    ///
50    /// Called once by the runtime, before the component runs, with the shutdown signal of the
51    /// component's dedicated supervisor.
52    pub(crate) fn set_shutdown_handle(&mut self, shutdown_handle: ShutdownHandle) {
53        self.shutdown_handle = Some(shutdown_handle);
54    }
55
56    /// Consumes the shutdown handle of this source context.
57    ///
58    /// # Panics
59    ///
60    /// Panics if the shutdown handle has already been taken.
61    pub fn take_shutdown_handle(&mut self) -> ShutdownHandle {
62        self.shutdown_handle.take().expect("shutdown handle already taken")
63    }
64
65    /// Consumes the health handle of this source context.
66    ///
67    /// # Panics
68    ///
69    /// Panics if the health handle has already been taken.
70    pub fn take_health_handle(&mut self) -> Health {
71        self.health_handle.take().expect("health handle already taken")
72    }
73
74    /// Gets a reference to the topology context.
75    pub fn topology_context(&self) -> &TopologyContext {
76        &self.inner.topology_context
77    }
78
79    /// Gets a reference to the component context.
80    pub fn component_context(&self) -> &ComponentContext {
81        &self.inner.component_context
82    }
83
84    /// Gets a reference to the component registry.
85    pub fn component_registry(&self) -> &ComponentRegistry {
86        &self.inner.component_registry
87    }
88
89    /// Gets a reference to the events dispatcher.
90    pub fn dispatcher(&self) -> &EventsDispatcher {
91        &self.inner.dispatcher
92    }
93
94    /// Returns a handle to the supervisor that this component is spawned on.
95    ///
96    /// Dynamic child processes can be spawned via the supervisor handle and thus have their lifecycle
97    /// coupled to the component itself: if the component restarts, or the component's supervisor dies,
98    /// the dynamic child processes will also be terminated automatically as well.
99    pub fn spawn_handle(&self) -> &SupervisorHandle {
100        &self.inner.supervisor_handle
101    }
102}
103
104impl Clone for SourceContext {
105    fn clone(&self) -> Self {
106        Self {
107            shutdown_handle: None,
108            health_handle: None,
109            inner: self.inner.clone(),
110        }
111    }
112}
saluki_core/components/sources/context.rs

saluki_core/components/sources/
context.rs
