reifydb_cdc/
testing.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::{mem, sync::Arc};
5
6use reifydb_catalog::{cache::CatalogCache, catalog::Catalog};
7use reifydb_core::{
8	common::CommitVersion,
9	encoded::{key::EncodedKey, row::EncodedRow},
10	event::EventBus,
11};
12use reifydb_runtime::{
13	actor::system::ActorSystem,
14	context::{
15		clock::{Clock, MockClock},
16		rng::Rng,
17	},
18	pool::Pools,
19};
20use reifydb_store_multi::MultiStore;
21use reifydb_store_single::SingleStore;
22use reifydb_transaction::{
23	interceptor::interceptors::Interceptors,
24	multi::transaction::MultiTransaction,
25	single::SingleTransaction,
26	transaction::{command::CommandTransaction, query::QueryTransaction},
27};
28use reifydb_value::{
29	Result,
30	util::cowvec::CowVec,
31	value::{duration::Duration, identity::IdentityId},
32};
33
34use crate::consume::host::CdcHost;
35
36#[derive(Clone)]
37pub struct TestCdcHost {
38	multi: MultiTransaction,
39	single: SingleTransaction,
40	pub event_bus: EventBus,
41	pub catalog: Catalog,
42	pub clock: Clock,
43	pub mock: MockClock,
44}
45
46impl TestCdcHost {
47	pub fn with_clock(initial_nanos: u64) -> Self {
48		let multi_store = MultiStore::testing_memory();
49		let single_store = SingleStore::testing_memory();
50		let actor_system = ActorSystem::new(Pools::default(), Clock::Real);
51		let spawner = actor_system.spawner();
52		mem::forget(actor_system);
53		let event_bus = EventBus::new(&spawner);
54		let single = SingleTransaction::new(single_store, event_bus.clone());
55		let catalog_cache = CatalogCache::new();
56		let mock = MockClock::new(initial_nanos);
57		let clock = Clock::Mock(mock.clone());
58		let multi = MultiTransaction::new(
59			multi_store,
60			single.clone(),
61			event_bus.clone(),
62			spawner,
63			clock.clone(),
64			Rng::seeded(42),
65			Arc::new(catalog_cache.clone()),
66		)
67		.unwrap();
68		Self {
69			multi,
70			single,
71			event_bus,
72			catalog: Catalog::new(catalog_cache),
73			clock,
74			mock,
75		}
76	}
77
78	pub fn new() -> Self {
79		Self::with_clock(1_000_000_000)
80	}
81}
82
83impl Default for TestCdcHost {
84	fn default() -> Self {
85		Self::new()
86	}
87}
88
89impl CdcHost for TestCdcHost {
90	fn begin_command(&self) -> Result<CommandTransaction> {
91		CommandTransaction::new(
92			self.multi.clone(),
93			self.single.clone(),
94			self.event_bus.clone(),
95			Interceptors::new(),
96			IdentityId::system(),
97			self.clock.clone(),
98		)
99	}
100
101	fn begin_query(&self) -> Result<QueryTransaction> {
102		Ok(QueryTransaction::new(self.multi.begin_query()?, self.single.clone(), IdentityId::system()))
103	}
104
105	fn current_version(&self) -> Result<CommitVersion> {
106		Ok(CommitVersion(1))
107	}
108
109	fn done_until(&self) -> CommitVersion {
110		CommitVersion(1)
111	}
112
113	fn cdc_producer_watermark(&self) -> CommitVersion {
114		CommitVersion(1)
115	}
116
117	fn wait_for_mark_timeout(&self, _version: CommitVersion, _timeout: Duration) -> bool {
118		true
119	}
120
121	fn notify_on_mark(&self, _version: CommitVersion, callback: Box<dyn FnOnce() + Send>) {
122		callback();
123	}
124
125	fn catalog(&self) -> &Catalog {
126		&self.catalog
127	}
128}
129
130pub fn make_key(s: &str) -> EncodedKey {
131	EncodedKey::new(s.as_bytes().to_vec())
132}
133
134pub fn make_row(s: &str) -> EncodedRow {
135	EncodedRow(CowVec::new(s.as_bytes().to_vec()))
136}
reifydb_cdc/testing.rs

reifydb_cdc/
testing.rs
