reifydb_sub_server/subscription/
host.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use reifydb_core::{
5	actors::server::Operation,
6	interface::catalog::subscription::{SubscribeOptions, SubscribeOutcome},
7};
8use reifydb_sub_core::host::{SubscribeContext, SubscribeHost};
9use reifydb_value::{params::Params, value::identity::IdentityId};
10
11use crate::{
12	dispatch::dispatch_subscribe,
13	execute::ExecuteError,
14	interceptor::{RequestContext, RequestMetadata},
15	state::AppState,
16};
17
18pub struct ServerSubscribeHost<'a> {
19	state: &'a AppState,
20	metadata: RequestMetadata,
21	context: SubscribeContext,
22}
23
24impl<'a> ServerSubscribeHost<'a> {
25	pub fn new(state: &'a AppState, metadata: RequestMetadata) -> Self {
26		let context = SubscribeContext::new(
27			state.engine_clone(),
28			state.clock().clone(),
29			state.rng().clone(),
30			state.subscribe_max_hydration_rows(),
31			state.subscribe_min_throttle(),
32			state.subscribe_min_linger(),
33		);
34		Self {
35			state,
36			metadata,
37			context,
38		}
39	}
40}
41
42impl SubscribeHost for ServerSubscribeHost<'_> {
43	type Error = ExecuteError;
44
45	fn context(&self) -> &SubscribeContext {
46		&self.context
47	}
48
49	async fn execute_subscribe(
50		&self,
51		identity: IdentityId,
52		query: String,
53		params: Params,
54		options: SubscribeOptions,
55	) -> Result<SubscribeOutcome, Self::Error> {
56		let ctx = RequestContext {
57			identity,
58			operation: Operation::Subscribe,
59			rql: query,
60			params,
61			metadata: self.metadata.clone(),
62		};
63		dispatch_subscribe(self.state, ctx, options).await
64	}
65}
66
67impl AppState {
68	pub fn subscribe_host(&self, metadata: RequestMetadata) -> ServerSubscribeHost<'_> {
69		ServerSubscribeHost::new(self, metadata)
70	}
71}
reifydb_sub_server/subscription/host.rs

reifydb_sub_server/subscription/
host.rs
