reifydb_cdc/consume/
poll.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::sync::{
5	Arc,
6	atomic::{AtomicBool, Ordering},
7};
8
9use reifydb_core::{
10	actors::cdc::{CdcPollHandle, CdcPollMessage},
11	interface::cdc::CdcConsumerId,
12};
13use reifydb_runtime::actor::system::ActorSpawner;
14use reifydb_value::{Result, value::duration::Duration};
15
16use super::{
17	actor::{PollActor, PollActorConfig},
18	consumer::{CdcConsume, CdcConsumer},
19	host::CdcHost,
20	wake::CdcWakeRegistry,
21	watermark::CdcConsumerWatermark,
22};
23use crate::storage::CdcStore;
24
25#[derive(Debug, Clone)]
26pub struct PollConsumerConfig {
27	pub consumer_id: CdcConsumerId,
28
29	pub thread_name: String,
30
31	pub poll_interval: Duration,
32
33	pub max_batch_size: Option<u64>,
34
35	pub consumer_watermark: Option<CdcConsumerWatermark>,
36
37	pub wake_registry: Option<CdcWakeRegistry>,
38}
39
40impl PollConsumerConfig {
41	pub fn new(
42		consumer_id: CdcConsumerId,
43		thread_name: impl Into<String>,
44		poll_interval: Duration,
45		max_batch_size: Option<u64>,
46	) -> Self {
47		Self {
48			consumer_id,
49			thread_name: thread_name.into(),
50			poll_interval,
51			max_batch_size,
52			consumer_watermark: None,
53			wake_registry: None,
54		}
55	}
56
57	pub fn with_consumer_watermark(mut self, watermark: CdcConsumerWatermark) -> Self {
58		self.consumer_watermark = Some(watermark);
59		self
60	}
61
62	pub fn with_wake_registry(mut self, registry: CdcWakeRegistry) -> Self {
63		self.wake_registry = Some(registry);
64		self
65	}
66}
67
68pub struct PollConsumer<H: CdcHost, C: CdcConsume + Send + 'static> {
69	config: PollConsumerConfig,
70	host: Option<H>,
71	consumer: Option<C>,
72	store: Option<CdcStore>,
73	running: Arc<AtomicBool>,
74	spawner: ActorSpawner,
75
76	handle: Option<CdcPollHandle>,
77}
78
79impl<H: CdcHost, C: CdcConsume + Send + 'static> PollConsumer<H, C> {
80	pub fn new(config: PollConsumerConfig, host: H, consume: C, store: CdcStore, spawner: ActorSpawner) -> Self {
81		Self {
82			config,
83			host: Some(host),
84			consumer: Some(consume),
85			store: Some(store),
86			running: Arc::new(AtomicBool::new(false)),
87			spawner,
88			handle: None,
89		}
90	}
91
92	fn take_resources(&mut self) -> (H, C, CdcStore) {
93		let host = self.host.take().expect("host already consumed");
94		let consumer = self.consumer.take().expect("consumer already consumed");
95		let store = self.store.take().expect("store already consumed");
96		(host, consumer, store)
97	}
98
99	fn build_actor_config(&self) -> PollActorConfig {
100		PollActorConfig {
101			consumer_id: self.config.consumer_id.clone(),
102			poll_interval: self.config.poll_interval,
103			max_batch_size: self.config.max_batch_size,
104		}
105	}
106}
107
108impl<H: CdcHost, C: CdcConsume + Send + Sync + 'static> CdcConsumer for PollConsumer<H, C> {
109	fn start(&mut self) -> Result<()> {
110		if self.running.swap(true, Ordering::AcqRel) {
111			return Ok(());
112		}
113		let (host, consumer, store) = self.take_resources();
114		let watermark = self.config.consumer_watermark.clone();
115		let wake_armed = Arc::new(AtomicBool::new(false));
116		let actor =
117			PollActor::new(self.build_actor_config(), host, consumer, store, watermark, wake_armed.clone());
118		let handle = self.spawner.spawn_system(&self.config.thread_name, actor);
119		if let Some(registry) = &self.config.wake_registry {
120			registry.register(wake_armed, handle.actor_ref().clone());
121		}
122		self.handle = Some(handle);
123		Ok(())
124	}
125
126	fn stop(&mut self) -> Result<()> {
127		if !self.running.swap(false, Ordering::AcqRel) {
128			return Ok(());
129		}
130
131		if let Some(handle) = self.handle.take() {
132			let _ = handle.actor_ref().send(CdcPollMessage::Shutdown);
133			let _ = handle.join();
134		}
135
136		Ok(())
137	}
138
139	fn is_running(&self) -> bool {
140		self.running.load(Ordering::Acquire)
141	}
142}
reifydb_cdc/consume/poll.rs

reifydb_cdc/consume/
poll.rs
