reifydb_sub_server/
state.rs1use std::time::Duration;
5
6use reifydb_auth::service::AuthService;
7use reifydb_core::actors::server::ServerMessage;
8use reifydb_engine::engine::StandardEngine;
9use reifydb_runtime::{
10 actor::{
11 mailbox::ActorRef,
12 system::{ActorHandle, ActorSystem},
13 },
14 context::{clock::Clock, rng::Rng},
15};
16use tracing::instrument;
17
18use crate::{actor::ServerActor, interceptor::RequestInterceptorChain};
19
20#[derive(Debug, Clone)]
21pub struct StateConfig {
22 pub query_timeout: Duration,
23
24 pub request_timeout: Duration,
25
26 pub max_connections: usize,
27
28 pub admin_enabled: bool,
29
30 pub subscribe_max_hydration_rows: u64,
31
32 pub subscribe_min_throttle: Duration,
33}
34
35impl Default for StateConfig {
36 fn default() -> Self {
37 Self {
38 query_timeout: Duration::from_secs(30),
39 request_timeout: Duration::from_secs(60),
40 max_connections: 10_000,
41 admin_enabled: false,
42 subscribe_max_hydration_rows: 10_000,
43 subscribe_min_throttle: Duration::from_millis(50),
44 }
45 }
46}
47
48impl StateConfig {
49 pub fn new() -> Self {
50 Self::default()
51 }
52
53 pub fn query_timeout(mut self, timeout: Duration) -> Self {
54 self.query_timeout = timeout;
55 self
56 }
57
58 pub fn request_timeout(mut self, timeout: Duration) -> Self {
59 self.request_timeout = timeout;
60 self
61 }
62
63 pub fn max_connections(mut self, max: usize) -> Self {
64 self.max_connections = max;
65 self
66 }
67
68 pub fn admin_enabled(mut self, enabled: bool) -> Self {
69 self.admin_enabled = enabled;
70 self
71 }
72
73 pub fn subscribe_max_hydration_rows(mut self, n: u64) -> Self {
74 self.subscribe_max_hydration_rows = n;
75 self
76 }
77
78 pub fn subscribe_min_throttle(mut self, interval: Duration) -> Self {
79 self.subscribe_min_throttle = interval;
80 self
81 }
82}
83
84#[derive(Clone)]
85pub struct AppState {
86 actor_system: ActorSystem,
87 engine: StandardEngine,
88 auth_service: AuthService,
89 config: StateConfig,
90 request_interceptors: RequestInterceptorChain,
91 clock: Clock,
92 rng: Rng,
93}
94
95impl AppState {
96 pub fn new(
97 actor_system: ActorSystem,
98 engine: StandardEngine,
99 auth_service: AuthService,
100 config: StateConfig,
101 request_interceptors: RequestInterceptorChain,
102 clock: Clock,
103 rng: Rng,
104 ) -> Self {
105 Self {
106 actor_system,
107 engine,
108 auth_service,
109 config,
110 request_interceptors,
111 clock,
112 rng,
113 }
114 }
115
116 pub fn clone_with_config(&self, config: StateConfig) -> Self {
117 Self {
118 actor_system: self.actor_system.clone(),
119 engine: self.engine.clone(),
120 auth_service: self.auth_service.clone(),
121 config,
122 request_interceptors: self.request_interceptors.clone(),
123 clock: self.clock.clone(),
124 rng: self.rng.clone(),
125 }
126 }
127
128 #[inline]
129 pub fn actor_system(&self) -> ActorSystem {
130 self.actor_system.clone()
131 }
132
133 #[inline]
134 pub fn engine(&self) -> &StandardEngine {
135 &self.engine
136 }
137
138 #[inline]
139 pub fn engine_clone(&self) -> StandardEngine {
140 self.engine.clone()
141 }
142
143 #[inline]
144 pub fn config(&self) -> &StateConfig {
145 &self.config
146 }
147
148 #[inline]
149 pub fn query_timeout(&self) -> Duration {
150 self.config.query_timeout
151 }
152
153 #[inline]
154 pub fn request_timeout(&self) -> Duration {
155 self.config.request_timeout
156 }
157
158 #[inline]
159 pub fn max_connections(&self) -> usize {
160 self.config.max_connections
161 }
162
163 #[inline]
164 pub fn admin_enabled(&self) -> bool {
165 self.config.admin_enabled
166 }
167
168 #[inline]
169 pub fn subscribe_max_hydration_rows(&self) -> u64 {
170 self.config.subscribe_max_hydration_rows
171 }
172
173 #[inline]
174 pub fn subscribe_min_throttle(&self) -> Duration {
175 self.config.subscribe_min_throttle
176 }
177
178 pub fn clamp_throttle(&self, requested: Option<Duration>) -> Duration {
179 match requested {
180 Some(d) if !d.is_zero() => d.max(self.config.subscribe_min_throttle),
181 _ => Duration::ZERO,
182 }
183 }
184
185 #[inline]
186 pub fn request_interceptors(&self) -> &RequestInterceptorChain {
187 &self.request_interceptors
188 }
189
190 #[inline]
191 pub fn clock(&self) -> &Clock {
192 &self.clock
193 }
194
195 #[inline]
196 pub fn rng(&self) -> &Rng {
197 &self.rng
198 }
199
200 #[inline]
201 pub fn auth_service(&self) -> &AuthService {
202 &self.auth_service
203 }
204
205 #[instrument(name = "actor::spawn_server", level = "debug", skip_all)]
206 pub fn spawn_server_actor(&self) -> (ActorRef<ServerMessage>, ActorHandle<ServerMessage>) {
207 let actor = ServerActor::new(self.engine.clone(), self.auth_service.clone(), self.clock.clone());
208 let handle = self.actor_system.spawn_query("server-req", actor);
209 let actor_ref = handle.actor_ref().clone();
210 (actor_ref, handle)
211 }
212}
213
214#[cfg(test)]
215pub mod tests {
216 use super::*;
217
218 #[test]
219 fn test_query_defaults() {
220 let config = StateConfig::default();
221 assert_eq!(config.query_timeout, Duration::from_secs(30));
222 assert_eq!(config.request_timeout, Duration::from_secs(60));
223 assert_eq!(config.max_connections, 10_000);
224 assert_eq!(config.subscribe_max_hydration_rows, 10_000);
225 assert_eq!(config.subscribe_min_throttle, Duration::from_millis(50));
226 }
227
228 #[test]
229 fn test_query_config_builder() {
230 let config = StateConfig::new()
231 .query_timeout(Duration::from_secs(60))
232 .request_timeout(Duration::from_secs(120))
233 .max_connections(5_000)
234 .subscribe_max_hydration_rows(50_000)
235 .subscribe_min_throttle(Duration::from_millis(200));
236
237 assert_eq!(config.query_timeout, Duration::from_secs(60));
238 assert_eq!(config.request_timeout, Duration::from_secs(120));
239 assert_eq!(config.max_connections, 5_000);
240 assert_eq!(config.subscribe_max_hydration_rows, 50_000);
241 assert_eq!(config.subscribe_min_throttle, Duration::from_millis(200));
242 }
243}